From cfc01bb3d962d0445d15a8cd7461cef8ec8d92c9 Mon Sep 17 00:00:00 2001 From: withchao <993506633@qq.com> Date: Fri, 31 May 2024 18:41:34 +0800 Subject: [PATCH] sync --- go.mod | 4 +- go.sum | 2 + internal/api/friend.go | 17 --- internal/api/group.go | 50 ++++++++ internal/api/router.go | 3 + internal/rpc/friend/sync.go | 114 +++++++++++------- internal/rpc/group/convert.go | 4 + internal/rpc/group/group.go | 61 +++++++--- internal/rpc/group/sync.go | 66 +++++++++- internal/rpc/incrversion/option.go | 58 +++++++++ pkg/common/config/config.go | 3 +- pkg/common/storage/cache/group.go | 3 + pkg/common/storage/cache/redis/group.go | 23 ++++ pkg/common/storage/controller/group.go | 55 +++++++-- pkg/common/storage/database/group_member.go | 3 + pkg/common/storage/database/mgo/friend.go | 10 +- .../storage/database/mgo/group_member.go | 95 ++++++++++++--- .../storage/database/mgo/version_log.go | 2 +- pkg/common/storage/database/version_log.go | 2 +- 19 files changed, 462 insertions(+), 113 deletions(-) create mode 100644 internal/rpc/incrversion/option.go diff --git a/go.mod b/go.mod index bd1e16a5b..db418bf14 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.2 + github.com/openimsdk/protocol v0.0.69-alpha.3 github.com/openimsdk/tools v0.0.49-alpha.24 github.com/pkg/errors v0.9.1 // indirect github.com/prometheus/client_golang v1.18.0 @@ -178,4 +178,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 1cc2cc1bc..4588df18a 100644 --- a/go.sum +++ b/go.sum @@ -286,6 +286,8 @@ github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE= github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs= 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.3 h1:Uf167FVB5EqYpiy2zBbR63OiK+Njjy99fXYasK6Zi+4= +github.com/openimsdk/protocol v0.0.69-alpha.3/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8= github.com/openimsdk/tools v0.0.49-alpha.24 h1:lJsqnjTPujnr91LRQ6QmcTliMIa4fMOBSTri6rFz2ek= github.com/openimsdk/tools v0.0.49-alpha.24/go.mod h1:g7mkHXYUPi0/8aAX8VPMHpnb3hqdV69Jph+bXOGvvNM= github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4= diff --git a/internal/api/friend.go b/internal/api/friend.go index 5b2dfa0ca..3af162a53 100644 --- a/internal/api/friend.go +++ b/internal/api/friend.go @@ -98,20 +98,3 @@ func (o *FriendApi) UpdateFriends(c *gin.Context) { func (o *FriendApi) GetIncrementalFriends(c *gin.Context) { a2r.Call(friend.FriendClient.GetIncrementalFriends, o.Client, c) } - -//func BatchIncremental[A, B, C any,D comparable](c *gin.Context, rpc func(client C, ctx context.Context, req *A, options ...grpc.CallOption) (*B, error), getID func(req *A)D, setID func(req *A, id D)) { -// req, err := a2r.ParseRequestNotCheck[BatchIncrementalReq[A]](c) -// if err != nil { -// apiresp.GinError(c, err) -// return -// } -// if len(req.List) == 0 { -// apiresp.GinError(c, errs.ErrArgs.WrapMsg("empty versions list")) -// return -// } -//} -// -//type BatchIncrementalReq[A any] struct { -// UserID string `json:"user_id"` -// List []*A `json:"list"` -//} diff --git a/internal/api/group.go b/internal/api/group.go index 6079c5343..7baa16323 100644 --- a/internal/api/group.go +++ b/internal/api/group.go @@ -19,6 +19,7 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/rpcclient" "github.com/openimsdk/protocol/group" "github.com/openimsdk/tools/a2r" + "github.com/openimsdk/tools/apiresp" ) type GroupApi rpcclient.Group @@ -134,3 +135,52 @@ func (o *GroupApi) GetGroups(c *gin.Context) { func (o *GroupApi) GetGroupMemberUserIDs(c *gin.Context) { a2r.Call(group.GroupClient.GetGroupMemberUserIDs, o.Client, c) } + +func (o *GroupApi) GetIncrementalJoinGroup(c *gin.Context) { + a2r.Call(group.GroupClient.GetIncrementalJoinGroup, o.Client, c) +} + +func (o *GroupApi) GetIncrementalGroupMember(c *gin.Context) { + a2r.Call(group.GroupClient.GetIncrementalGroupMember, o.Client, c) +} + +func (o *GroupApi) GetIncrementalGroupMemberBatch(c *gin.Context) { + type BatchIncrementalReq[A any] struct { + UserID string `json:"user_id"` + List []*group.GetIncrementalGroupMemberReq `json:"list"` + } + type BatchIncrementalResp struct { + List map[string]*group.GetIncrementalGroupMemberResp `json:"list"` + } + req, err := a2r.ParseRequestNotCheck[BatchIncrementalReq](c) + if err != nil { + apiresp.GinError(c, err) + return + } + resp := &BatchIncrementalResp{ + List: make(map[string]*group.GetIncrementalGroupMemberResp), + } + var ( + changeCount int + ) + for _, req := range req.List { + if _, ok := resp.List[req.GroupID]; ok { + continue + } + res, err := o.Client.GetIncrementalGroupMember(c, req) + if err != nil { + if len(resp.List) == 0 { + apiresp.GinError(c, err) + } else { + apiresp.GinSuccess(c, resp) + } + return + } + resp.List[req.GroupID] = res + changeCount += len(res.Changes) + len(res.DeleteUserIds) + if changeCount > int(res.SyncCount)*4 { + break + } + } + apiresp.GinSuccess(c, resp) +} diff --git a/internal/api/router.go b/internal/api/router.go index 78e049e0f..8f2733604 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -115,6 +115,9 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En groupRouterGroup.POST("/get_group_abstract_info", g.GetGroupAbstractInfo) groupRouterGroup.POST("/get_groups", g.GetGroups) groupRouterGroup.POST("/get_group_member_user_id", g.GetGroupMemberUserIDs) + groupRouterGroup.POST("/get_incremental_join_group", g.GetIncrementalJoinGroup) + groupRouterGroup.POST("/get_incremental_group_member", g.GetIncrementalGroupMember) + groupRouterGroup.POST("/get_incremental_group_member_batch", g.GetIncrementalGroupMemberBatch) } // certificate authRouterGroup := r.Group("/auth") diff --git a/internal/rpc/friend/sync.go b/internal/rpc/friend/sync.go index bfa7e3a57..bdea7b9eb 100644 --- a/internal/rpc/friend/sync.go +++ b/internal/rpc/friend/sync.go @@ -2,9 +2,7 @@ package friend import ( "context" - "crypto/md5" - "encoding/binary" - "encoding/json" + "github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" "github.com/openimsdk/open-im-server/v3/pkg/authverify" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" pbfriend "github.com/openimsdk/protocol/friend" @@ -59,49 +57,79 @@ func (s *friendServer) SearchFriends(ctx context.Context, req *pbfriend.SearchFr }, nil } -func (s *friendServer) sortFriendUserIDsHash(userIDs []string) uint64 { - data, _ := json.Marshal(userIDs) - sum := md5.Sum(data) - return binary.BigEndian.Uint64(sum[:]) -} - func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *pbfriend.GetIncrementalFriendsReq) (*pbfriend.GetIncrementalFriendsResp, error) { if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil { return nil, err } - var limit int - if req.Version > 0 { - limit = s.config.RpcConfig.FriendSyncCount - } - incrVer, err := s.friendDatabase.FindFriendIncrVersion(ctx, req.UserID, uint(req.Version), limit) - if err != nil { - return nil, err + opt := incrversion.Option[*pbfriend.FriendInfo, pbfriend.GetIncrementalFriendsResp]{ + VersionID: req.VersionID, + Version: func() (*model.VersionLog, error) { + return s.friendDatabase.FindFriendIncrVersion(ctx, req.UserID, uint(req.Version), incrversion.Limit(s.config.RpcConfig.FriendSyncCount, req.Version)) + }, + AllID: func() ([]string, error) { + return s.friendDatabase.FindSortFriendUserIDs(ctx, req.UserID) + }, + Find: func(ids []string) ([]*pbfriend.FriendInfo, error) { + friends, err := s.friendDatabase.FindFriendsWithError(ctx, req.UserID, ids) + if err != nil { + return nil, err + } + return friendsDB2PB(friends), nil + }, + ID: func(elem *pbfriend.FriendInfo) string { + return elem.FriendUserID + }, + Resp: func(version *model.VersionLog, delIDs []string, list []*pbfriend.FriendInfo, full bool) *pbfriend.GetIncrementalFriendsResp { + return &pbfriend.GetIncrementalFriendsResp{ + VersionID: version.ID.Hex(), + Version: uint64(version.Version), + Full: full, + SyncCount: uint32(s.config.RpcConfig.FriendSyncCount), + DeleteUserIds: delIDs, + Changes: list, + } + }, } - var ( - deleteUserIDs []string - changeUserIDs []string - ) - if incrVer.Full() { - changeUserIDs, err = s.friendDatabase.FindSortFriendUserIDs(ctx, req.UserID) - if err != nil { - return nil, err - } - } else { - deleteUserIDs, changeUserIDs = incrVer.DeleteAndChangeIDs() - } - var friends []*model.Friend - if len(changeUserIDs) > 0 { - friends, err = s.friendDatabase.FindFriendsWithError(ctx, req.UserID, changeUserIDs) - if err != nil { - return nil, err - } - } - return &pbfriend.GetIncrementalFriendsResp{ - Version: uint64(incrVer.Version), - VersionID: incrVer.ID.Hex(), - Full: incrVer.Full(), - SyncCount: uint32(s.config.RpcConfig.FriendSyncCount), - DeleteUserIds: deleteUserIDs, - Changes: friendsDB2PB(friends), - }, nil + return opt.Build() } + +//func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *pbfriend.GetIncrementalFriendsReq) (*pbfriend.GetIncrementalFriendsResp, error) { +// if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil { +// return nil, err +// } +// var limit int +// if req.Version > 0 { +// limit = s.config.RpcConfig.FriendSyncCount +// } +// incrVer, err := s.friendDatabase.FindFriendIncrVersion(ctx, req.UserID, uint(req.Version), limit) +// if err != nil { +// return nil, err +// } +// var ( +// deleteUserIDs []string +// changeUserIDs []string +// ) +// if incrVer.Full() { +// changeUserIDs, err = s.friendDatabase.FindSortFriendUserIDs(ctx, req.UserID) +// if err != nil { +// return nil, err +// } +// } else { +// deleteUserIDs, changeUserIDs = incrVer.DeleteAndChangeIDs() +// } +// var friends []*model.Friend +// if len(changeUserIDs) > 0 { +// friends, err = s.friendDatabase.FindFriendsWithError(ctx, req.UserID, changeUserIDs) +// if err != nil { +// return nil, err +// } +// } +// return &pbfriend.GetIncrementalFriendsResp{ +// Version: uint64(incrVer.Version), +// VersionID: incrVer.ID.Hex(), +// Full: incrVer.Full(), +// SyncCount: uint32(s.config.RpcConfig.FriendSyncCount), +// DeleteUserIds: deleteUserIDs, +// Changes: friendsDB2PB(friends), +// }, nil +//} diff --git a/internal/rpc/group/convert.go b/internal/rpc/group/convert.go index a75693904..8026430c3 100644 --- a/internal/rpc/group/convert.go +++ b/internal/rpc/group/convert.go @@ -57,3 +57,7 @@ func (s *groupServer) groupMemberDB2PB(member *model.GroupMember, appMangerLevel InviterUserID: member.InviterUserID, } } + +func (s *groupServer) groupMemberDB2PB2(member *model.GroupMember) *sdkws.GroupMemberFullInfo { + return s.groupMemberDB2PB(member, 0) +} diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index 51fd2d7b6..9b9ef07c1 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -77,6 +77,9 @@ type Config struct { } func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error { + if config.RpcConfig.GroupSyncCount <= 0 { + config.RpcConfig.GroupSyncCount = constant.MaxSyncPullNumber + } mgocli, err := mongoutil.NewMongoDB(ctx, config.MongodbConfig.Build()) if err != nil { return err @@ -642,18 +645,29 @@ func (s *groupServer) GetGroupMembersInfo(ctx context.Context, req *pbgroup.GetG if req.GroupID == "" { return nil, errs.ErrArgs.WrapMsg("groupID empty") } - members, err := s.db.FindGroupMembers(ctx, req.GroupID, req.UserIDs) + members, err := s.getGroupMembersInfo(ctx, req.GroupID, req.UserIDs) + if err != nil { + return nil, err + } + return &pbgroup.GetGroupMembersInfoResp{ + Members: members, + }, nil +} + +func (s *groupServer) getGroupMembersInfo(ctx context.Context, groupID string, userIDs []string) ([]*sdkws.GroupMemberFullInfo, error) { + if len(userIDs) == 0 { + return nil, nil + } + members, err := s.db.FindGroupMembers(ctx, groupID, userIDs) if err != nil { return nil, err } if err := s.PopulateGroupMember(ctx, members...); err != nil { return nil, err } - return &pbgroup.GetGroupMembersInfoResp{ - Members: datautil.Slice(members, func(e *model.GroupMember) *sdkws.GroupMemberFullInfo { - return convert.Db2PbGroupMember(e) - }), - }, nil + return datautil.Slice(members, func(e *model.GroupMember) *sdkws.GroupMemberFullInfo { + return convert.Db2PbGroupMember(e) + }), nil } // GetGroupApplicationList handles functions that get a list of group requests. @@ -722,15 +736,28 @@ func (s *groupServer) GetGroupsInfo(ctx context.Context, req *pbgroup.GetGroupsI if len(req.GroupIDs) == 0 { return nil, errs.ErrArgs.WrapMsg("groupID is empty") } - groups, err := s.db.FindGroup(ctx, req.GroupIDs) + groups, err := s.getGroupsInfo(ctx, req.GroupIDs) + if err != nil { + return nil, err + } + return &pbgroup.GetGroupsInfoResp{ + GroupInfos: groups, + }, nil +} + +func (s *groupServer) getGroupsInfo(ctx context.Context, groupIDs []string) ([]*sdkws.GroupInfo, error) { + if len(groupIDs) == 0 { + return nil, nil + } + groups, err := s.db.FindGroup(ctx, groupIDs) if err != nil { return nil, err } - groupMemberNumMap, err := s.db.MapGroupMemberNum(ctx, req.GroupIDs) + groupMemberNumMap, err := s.db.MapGroupMemberNum(ctx, groupIDs) if err != nil { return nil, err } - owners, err := s.db.FindGroupsOwner(ctx, req.GroupIDs) + owners, err := s.db.FindGroupsOwner(ctx, groupIDs) if err != nil { return nil, err } @@ -740,15 +767,13 @@ func (s *groupServer) GetGroupsInfo(ctx context.Context, req *pbgroup.GetGroupsI ownerMap := datautil.SliceToMap(owners, func(e *model.GroupMember) string { return e.GroupID }) - return &pbgroup.GetGroupsInfoResp{ - GroupInfos: datautil.Slice(groups, func(e *model.Group) *sdkws.GroupInfo { - var ownerUserID string - if owner, ok := ownerMap[e.GroupID]; ok { - ownerUserID = owner.UserID - } - return convert.Db2PbGroupInfo(e, ownerUserID, groupMemberNumMap[e.GroupID]) - }), - }, nil + return datautil.Slice(groups, func(e *model.Group) *sdkws.GroupInfo { + var ownerUserID string + if owner, ok := ownerMap[e.GroupID]; ok { + ownerUserID = owner.UserID + } + return convert.Db2PbGroupInfo(e, ownerUserID, groupMemberNumMap[e.GroupID]) + }), nil } func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbgroup.GroupApplicationResponseReq) (*pbgroup.GroupApplicationResponseResp, error) { diff --git a/internal/rpc/group/sync.go b/internal/rpc/group/sync.go index 765bceb63..80cb5a9ea 100644 --- a/internal/rpc/group/sync.go +++ b/internal/rpc/group/sync.go @@ -2,7 +2,11 @@ package group import ( "context" + "github.com/openimsdk/open-im-server/v3/internal/rpc/incrversion" + "github.com/openimsdk/open-im-server/v3/pkg/authverify" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" pbgroup "github.com/openimsdk/protocol/group" + "github.com/openimsdk/protocol/sdkws" ) func (s *groupServer) SearchGroupMember(ctx context.Context, req *pbgroup.SearchGroupMemberReq) (*pbgroup.SearchGroupMemberResp, error) { @@ -10,7 +14,63 @@ func (s *groupServer) SearchGroupMember(ctx context.Context, req *pbgroup.Search panic("implement me") } -func (s *groupServer) GetGroupMemberHash(ctx context.Context, req *pbgroup.GetGroupMemberHashReq) (*pbgroup.GetGroupMemberHashResp, error) { - //TODO implement me - panic("implement me") +func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgroup.GetIncrementalGroupMemberReq) (*pbgroup.GetIncrementalGroupMemberResp, error) { + opt := incrversion.Option[*sdkws.GroupMemberFullInfo, pbgroup.GetIncrementalGroupMemberResp]{ + VersionID: req.VersionID, + Version: func() (*model.VersionLog, error) { + return s.db.FindMemberIncrVersion(ctx, req.GroupID, uint(req.Version), incrversion.Limit(s.config.RpcConfig.GroupSyncCount, req.Version)) + }, + AllID: func() ([]string, error) { + return s.db.FindSortGroupMemberUserIDs(ctx, req.GroupID) + }, + Find: func(ids []string) ([]*sdkws.GroupMemberFullInfo, error) { + return s.getGroupMembersInfo(ctx, req.GroupID, ids) + }, + ID: func(elem *sdkws.GroupMemberFullInfo) string { + return elem.UserID + }, + Resp: func(version *model.VersionLog, delIDs []string, list []*sdkws.GroupMemberFullInfo, full bool) *pbgroup.GetIncrementalGroupMemberResp { + return &pbgroup.GetIncrementalGroupMemberResp{ + VersionID: version.ID.Hex(), + Version: uint64(version.Version), + Full: full, + SyncCount: uint32(s.config.RpcConfig.GroupSyncCount), + DeleteUserIds: delIDs, + Changes: list, + } + }, + } + return opt.Build() +} + +func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupReq) (*pbgroup.GetIncrementalJoinGroupResp, error) { + if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil { + return nil, err + } + opt := incrversion.Option[*sdkws.GroupInfo, pbgroup.GetIncrementalJoinGroupResp]{ + VersionID: req.VersionID, + Version: func() (*model.VersionLog, error) { + return s.db.FindJoinIncrVersion(ctx, req.UserID, uint(req.Version), incrversion.Limit(s.config.RpcConfig.GroupSyncCount, req.Version)) + }, + AllID: func() ([]string, error) { + return s.db.FindSortJoinGroupIDs(ctx, req.UserID) + }, + Find: func(ids []string) ([]*sdkws.GroupInfo, error) { + return s.getGroupsInfo(ctx, ids) + }, + ID: func(elem *sdkws.GroupInfo) string { + return elem.GroupID + }, + Resp: func(version *model.VersionLog, delIDs []string, list []*sdkws.GroupInfo, full bool) *pbgroup.GetIncrementalJoinGroupResp { + return &pbgroup.GetIncrementalJoinGroupResp{ + VersionID: version.ID.Hex(), + Version: uint64(version.Version), + Full: full, + SyncCount: uint32(s.config.RpcConfig.GroupSyncCount), + DeleteGroupIds: delIDs, + Changes: list, + } + }, + } + return opt.Build() } diff --git a/internal/rpc/incrversion/option.go b/internal/rpc/incrversion/option.go new file mode 100644 index 000000000..3e19056bf --- /dev/null +++ b/internal/rpc/incrversion/option.go @@ -0,0 +1,58 @@ +package incrversion + +import ( + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" + "github.com/openimsdk/tools/utils/datautil" +) + +func Limit(maxSync int, version uint64) int { + if version == 0 { + return 0 + } + return maxSync +} + +type Option[A, B any] struct { + VersionID string + Version func() (*model.VersionLog, error) + AllID func() ([]string, error) + Find func(ids []string) ([]A, error) + ID func(elem A) string + Resp func(version *model.VersionLog, delIDs []string, list []A, full bool) *B +} + +func (o *Option[A, B]) Build() (*B, error) { + version, err := o.Version() + if err != nil { + return nil, err + } + var ( + deleteIDs []string + changeIDs []string + ) + full := o.VersionID != version.ID.Hex() || version.Full() + if full { + changeIDs, err = o.AllID() + if err != nil { + return nil, err + } + } else { + deleteIDs, changeIDs = version.DeleteAndChangeIDs() + } + var list []A + if len(changeIDs) > 0 { + list, err = o.Find(changeIDs) + if err != nil { + return nil, err + } + if (!full) && o.ID != nil && len(changeIDs) != len(list) { + foundIDs := datautil.SliceSetAny(list, o.ID) + for _, id := range changeIDs { + if _, ok := foundIDs[id]; !ok { + deleteIDs = append(deleteIDs, id) + } + } + } + } + return o.Resp(version, deleteIDs, list, full), nil +} diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index da62acebf..31a54c371 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -254,7 +254,8 @@ type Group struct { ListenIP string `mapstructure:"listenIP"` Ports []int `mapstructure:"ports"` } `mapstructure:"rpc"` - Prometheus Prometheus `mapstructure:"prometheus"` + Prometheus Prometheus `mapstructure:"prometheus"` + GroupSyncCount int `mapstructure:"groupSyncCount"` } type Msg struct { diff --git a/pkg/common/storage/cache/group.go b/pkg/common/storage/cache/group.go index 53e2cd1c7..ff840a4a5 100644 --- a/pkg/common/storage/cache/group.go +++ b/pkg/common/storage/cache/group.go @@ -59,4 +59,7 @@ type GroupCache interface { GetGroupRolesLevelMemberInfo(ctx context.Context, groupID string, roleLevels []int32) ([]*model.GroupMember, error) GetGroupMemberNum(ctx context.Context, groupID string) (memberNum int64, err error) DelGroupsMemberNum(groupID ...string) GroupCache + + FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) + FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) } diff --git a/pkg/common/storage/cache/redis/group.go b/pkg/common/storage/cache/redis/group.go index 2de03906f..7f0ba62e4 100644 --- a/pkg/common/storage/cache/redis/group.go +++ b/pkg/common/storage/cache/redis/group.go @@ -46,6 +46,7 @@ type GroupCacheRedis struct { expireTime time.Duration rcClient *rockscache.Client groupHash cache.GroupHash + syncCount int } func NewGroupCacheRedis( @@ -406,3 +407,25 @@ func (g *GroupCacheRedis) FindGroupMemberUser(ctx context.Context, groupIDs []st return g.groupMemberDB.Take(ctx, groupID, userID) }) } + +func (g *GroupCacheRedis) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) { + userIDs, err := g.GetGroupMemberIDs(ctx, groupID) + if err != nil { + return nil, err + } + if len(userIDs) > g.syncCount { + userIDs = userIDs[:g.syncCount] + } + return userIDs, nil +} + +func (g *GroupCacheRedis) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) { + groupIDs, err := g.GetJoinedGroupIDs(ctx, userID) + if err != nil { + return nil, err + } + if len(groupIDs) > g.syncCount { + groupIDs = groupIDs[:g.syncCount] + } + return groupIDs, nil +} diff --git a/pkg/common/storage/controller/group.go b/pkg/common/storage/controller/group.go index f2a135835..c32f009dd 100644 --- a/pkg/common/storage/controller/group.go +++ b/pkg/common/storage/controller/group.go @@ -106,6 +106,13 @@ type GroupDatabase interface { CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error) // DeleteGroupMemberHash deletes the hash entries for group members in specified groups. DeleteGroupMemberHash(ctx context.Context, groupIDs []string) error + + 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) + + FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) + + FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) } func NewGroupDatabase( @@ -219,10 +226,21 @@ func (g *groupDatabase) SearchGroup(ctx context.Context, keyword string, paginat } func (g *groupDatabase) UpdateGroup(ctx context.Context, groupID string, data map[string]any) error { - if err := g.groupDB.UpdateMap(ctx, groupID, data); err != nil { - return err - } - return g.cache.DelGroupsInfo(groupID).ChainExecDel(ctx) + return g.ctxTx.Transaction(ctx, func(ctx context.Context) error { + if err := g.groupDB.UpdateMap(ctx, groupID, data); err != nil { + return err + } + userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) + if err != nil { + return err + } + for _, userID := range userIDs { + if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, userID, []string{groupID}, false); err != nil { + return err + } + } + return g.cache.DelGroupsInfo(groupID).ChainExecDel(ctx) + }) } func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, deleteMember bool) error { @@ -231,11 +249,11 @@ func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, delete if err := g.groupDB.UpdateStatus(ctx, groupID, constant.GroupStatusDismissed); err != nil { return err } + userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) + if err != nil { + return err + } if deleteMember { - userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID) - if err != nil { - return err - } if err := g.groupMemberDB.Delete(ctx, groupID, nil); err != nil { return err } @@ -246,6 +264,11 @@ func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, delete DelGroupAllRoleLevel(groupID). DelGroupMembersInfo(groupID, userIDs...) } + if len(userIDs) > 0 { + if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, groupID, userIDs, true); err != nil { + return err + } + } return c.DelGroupsInfo(groupID).ChainExecDel(ctx) }) } @@ -443,3 +466,19 @@ func (g *groupDatabase) DeleteGroupMemberHash(ctx context.Context, groupIDs []st } return c.ChainExecDel(ctx) } + +func (g *groupDatabase) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) { + return g.groupMemberDB.FindMemberIncrVersion(ctx, groupID, version, limit) +} + +func (g *groupDatabase) FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error) { + return g.groupMemberDB.FindJoinIncrVersion(ctx, userID, version, limit) +} + +func (g *groupDatabase) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) { + return g.cache.FindSortGroupMemberUserIDs(ctx, groupID) +} + +func (g *groupDatabase) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) { + return g.cache.FindSortJoinGroupIDs(ctx, userID) +} diff --git a/pkg/common/storage/database/group_member.go b/pkg/common/storage/database/group_member.go index f57f2c317..c397eda58 100644 --- a/pkg/common/storage/database/group_member.go +++ b/pkg/common/storage/database/group_member.go @@ -34,4 +34,7 @@ type GroupMember interface { TakeGroupMemberNum(ctx context.Context, groupID string) (count int64, err error) FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) IsUpdateRoleLevel(data map[string]any) bool + JoinGroupIncrVersion(ctx context.Context, userID string, groupIDs []string, deleted bool) error + 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) } diff --git a/pkg/common/storage/database/mgo/friend.go b/pkg/common/storage/database/mgo/friend.go index 2df85a430..83daf9ecd 100644 --- a/pkg/common/storage/database/mgo/friend.go +++ b/pkg/common/storage/database/mgo/friend.go @@ -45,7 +45,7 @@ func NewFriendMongo(db *mongo.Database) (database.Friend, error) { if err != nil { return nil, err } - owner, err := NewVersionLog(db.Collection("friend_owner_version_log")) + owner, err := NewVersionLog(db.Collection("friend_version")) if err != nil { return nil, err } @@ -62,7 +62,7 @@ func (f *FriendMgo) Create(ctx context.Context, friends []*model.Friend) error { mp[friend.OwnerUserID] = append(mp[friend.OwnerUserID], friend.FriendUserID) } for ownerUserID, friendUserIDs := range mp { - if err := f.owner.WriteLog(ctx, ownerUserID, friendUserIDs, false); err != nil { + if err := f.owner.IncrVersion(ctx, ownerUserID, friendUserIDs, false); err != nil { return err } } @@ -79,7 +79,7 @@ func (f *FriendMgo) Delete(ctx context.Context, ownerUserID string, friendUserID return mongoutil.IncrVersion(func() error { return mongoutil.DeleteOne(ctx, f.coll, filter) }, func() error { - return f.owner.WriteLog(ctx, ownerUserID, friendUserIDs, true) + return f.owner.IncrVersion(ctx, ownerUserID, friendUserIDs, true) }) } @@ -95,7 +95,7 @@ func (f *FriendMgo) UpdateByMap(ctx context.Context, ownerUserID string, friendU return mongoutil.IncrVersion(func() error { return mongoutil.UpdateOne(ctx, f.coll, filter, bson.M{"$set": args}, true) }, func() error { - return f.owner.WriteLog(ctx, ownerUserID, []string{friendUserID}, false) + return f.owner.IncrVersion(ctx, ownerUserID, []string{friendUserID}, false) }) } @@ -185,7 +185,7 @@ func (f *FriendMgo) UpdateFriends(ctx context.Context, ownerUserID string, frien return mongoutil.IncrVersion(func() error { return mongoutil.Ignore(mongoutil.UpdateMany(ctx, f.coll, filter, update)) }, func() error { - return f.owner.WriteLog(ctx, ownerUserID, friendUserIDs, false) + return f.owner.IncrVersion(ctx, ownerUserID, friendUserIDs, false) }) } diff --git a/pkg/common/storage/database/mgo/group_member.go b/pkg/common/storage/database/mgo/group_member.go index ccca386e5..372babc9f 100644 --- a/pkg/common/storage/database/mgo/group_member.go +++ b/pkg/common/storage/database/mgo/group_member.go @@ -40,15 +40,53 @@ func NewGroupMember(db *mongo.Database) (database.GroupMember, error) { if err != nil { return nil, errs.Wrap(err) } - return &GroupMemberMgo{coll: coll}, nil + member, err := NewVersionLog(db.Collection("group_member_version")) + if err != nil { + return nil, err + } + join, err := NewVersionLog(db.Collection("group_join_version")) + if err != nil { + return nil, err + } + return &GroupMemberMgo{coll: coll, member: member, join: join}, nil } type GroupMemberMgo struct { - coll *mongo.Collection + coll *mongo.Collection + member database.VersionLog + join database.VersionLog +} + +func (g *GroupMemberMgo) sortBson() any { + return bson.D{{"role_level", -1}, {"create_time", -1}} } func (g *GroupMemberMgo) Create(ctx context.Context, groupMembers []*model.GroupMember) (err error) { - return mongoutil.InsertMany(ctx, g.coll, groupMembers) + return mongoutil.IncrVersion(func() error { + return mongoutil.InsertMany(ctx, g.coll, groupMembers) + }, func() error { + gms := make(map[string][]string) + for _, member := range groupMembers { + gms[member.GroupID] = append(gms[member.GroupID], member.UserID) + } + for groupID, userIDs := range gms { + if err := g.member.IncrVersion(ctx, groupID, userIDs, false); err != nil { + return err + } + } + return nil + }, func() error { + gms := make(map[string][]string) + for _, member := range groupMembers { + gms[member.UserID] = append(gms[member.UserID], member.GroupID) + } + for userID, groupIDs := range gms { + if err := g.join.IncrVersion(ctx, userID, groupIDs, false); err != nil { + return err + } + } + return nil + }) } func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []string) (err error) { @@ -56,24 +94,41 @@ func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []s if len(userIDs) > 0 { filter["user_id"] = bson.M{"$in": userIDs} } - return mongoutil.DeleteMany(ctx, g.coll, filter) + return mongoutil.IncrVersion(func() error { + return mongoutil.DeleteMany(ctx, g.coll, filter) + }, func() error { + return g.member.IncrVersion(ctx, groupID, userIDs, true) + }, func() error { + if len(userIDs) == 0 { + return nil + } + return g.member.IncrVersion(ctx, groupID, userIDs, true) + }) } func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, userID string, roleLevel int32) error { - return g.Update(ctx, groupID, userID, bson.M{"role_level": roleLevel}) + return mongoutil.IncrVersion(func() error { + return g.Update(ctx, groupID, userID, bson.M{"role_level": roleLevel}) + }, func() error { + return g.member.IncrVersion(ctx, groupID, []string{userID}, true) + }, func() error { + return g.join.IncrVersion(ctx, groupID, []string{userID}, true) + }) } func (g *GroupMemberMgo) Update(ctx context.Context, groupID string, userID string, data map[string]any) (err error) { - return mongoutil.UpdateOne(ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID}, bson.M{"$set": data}, true) -} - -func (g *GroupMemberMgo) Find(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32) (groupMembers []*model.GroupMember, err error) { - // TODO implement me - panic("implement me") + if len(data) == 0 { + return nil + } + return mongoutil.IncrVersion(func() error { + return mongoutil.UpdateOne(ctx, g.coll, bson.M{"group_id": groupID, "user_id": userID}, bson.M{"$set": data}, true) + }, func() error { + return g.member.IncrVersion(ctx, groupID, []string{userID}, false) + }) } func (g *GroupMemberMgo) FindMemberUserID(ctx context.Context, groupID string) (userIDs []string, err error) { - return mongoutil.Find[string](ctx, g.coll, bson.M{"group_id": groupID}, options.Find().SetProjection(bson.M{"_id": 0, "user_id": 1})) + return mongoutil.Find[string](ctx, g.coll, bson.M{"group_id": groupID}, options.Find().SetProjection(bson.M{"_id": 0, "user_id": 1}).SetSort(g.sortBson())) } func (g *GroupMemberMgo) Take(ctx context.Context, groupID string, userID string) (groupMember *model.GroupMember, err error) { @@ -90,11 +145,11 @@ func (g *GroupMemberMgo) FindRoleLevelUserIDs(ctx context.Context, groupID strin func (g *GroupMemberMgo) SearchMember(ctx context.Context, keyword string, groupID string, pagination pagination.Pagination) (total int64, groupList []*model.GroupMember, err error) { filter := bson.M{"group_id": groupID, "nickname": bson.M{"$regex": keyword}} - return mongoutil.FindPage[*model.GroupMember](ctx, g.coll, filter, pagination) + return mongoutil.FindPage[*model.GroupMember](ctx, g.coll, filter, pagination, options.Find().SetSort(g.sortBson())) } func (g *GroupMemberMgo) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) { - return mongoutil.Find[string](ctx, g.coll, bson.M{"user_id": userID}, options.Find().SetProjection(bson.M{"_id": 0, "group_id": 1})) + return mongoutil.Find[string](ctx, g.coll, bson.M{"user_id": userID}, options.Find().SetProjection(bson.M{"_id": 0, "group_id": 1}).SetSort(g.sortBson())) } func (g *GroupMemberMgo) TakeGroupMemberNum(ctx context.Context, groupID string) (count int64, err error) { @@ -118,3 +173,15 @@ func (g *GroupMemberMgo) IsUpdateRoleLevel(data map[string]any) bool { _, ok := data["role_level"] return ok } + +func (g *GroupMemberMgo) JoinGroupIncrVersion(ctx context.Context, userID string, groupIDs []string, deleted bool) error { + return g.join.IncrVersion(ctx, userID, groupIDs, deleted) +} + +func (g *GroupMemberMgo) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) { + return g.member.FindChangeLog(ctx, groupID, version, limit) +} + +func (g *GroupMemberMgo) FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error) { + 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 fe123073e..5629c5c00 100644 --- a/pkg/common/storage/database/mgo/version_log.go +++ b/pkg/common/storage/database/mgo/version_log.go @@ -36,7 +36,7 @@ func (l *VersionLogMgo) initIndex(ctx context.Context) error { return err } -func (l *VersionLogMgo) WriteLog(ctx context.Context, dId string, eIds []string, deleted bool) error { +func (l *VersionLogMgo) IncrVersion(ctx context.Context, dId string, eIds []string, deleted bool) error { if len(eIds) == 0 { return errs.ErrArgs.WrapMsg("elem id is empty", "dId", dId) } diff --git a/pkg/common/storage/database/version_log.go b/pkg/common/storage/database/version_log.go index 6783cf4ff..3450d2776 100644 --- a/pkg/common/storage/database/version_log.go +++ b/pkg/common/storage/database/version_log.go @@ -12,7 +12,7 @@ const ( ) type VersionLog interface { - WriteLog(ctx context.Context, dId string, eIds []string, deleted bool) error + IncrVersion(ctx context.Context, dId string, eIds []string, deleted bool) error FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error) DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error }