blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-12-21 14:27:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

sync

Browse Source
This commit is contained in:
withchao 2024-05-31 18:41:34 +08:00
parent 0f72de85b2
commit cfc01bb3d9
19 changed files with 462 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
go.mod
View File

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

2
go.sum
View File

@ -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=

17
internal/api/friend.go
View File

@ -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"`
//}

50
internal/api/group.go
View File

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

3
internal/api/router.go
View File

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

106
internal/rpc/friend/sync.go
View File

@ -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)
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
}
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 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{
Version: uint64(incrVer.Version),
VersionID: incrVer.ID.Hex(),
Full: incrVer.Full(),
VersionID: version.ID.Hex(),
Version: uint64(version.Version),
Full: full,
SyncCount: uint32(s.config.RpcConfig.FriendSyncCount),
DeleteUserIds: deleteUserIDs,
Changes: friendsDB2PB(friends),
}, nil
DeleteUserIds: delIDs,
Changes: list,
}
},
}
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
//}

4
internal/rpc/group/convert.go
View File

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

49
internal/rpc/group/group.go
View File

@ -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 datautil.Slice(members, func(e *model.GroupMember) *sdkws.GroupMemberFullInfo {
return convert.Db2PbGroupMember(e)
}),
}, nil
}), 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
}
groupMemberNumMap, err := s.db.MapGroupMemberNum(ctx, req.GroupIDs)
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
}
owners, err := s.db.FindGroupsOwner(ctx, req.GroupIDs)
groupMemberNumMap, err := s.db.MapGroupMemberNum(ctx, groupIDs)
if err != nil {
return nil, err
}
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 {
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
}), nil
}
func (s *groupServer) GroupApplicationResponse(ctx context.Context, req *pbgroup.GroupApplicationResponseReq) (*pbgroup.GroupApplicationResponseResp, error) {

66
internal/rpc/group/sync.go
View File

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

58
internal/rpc/incrversion/option.go Normal file
View File

@ -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
}

1
pkg/common/config/config.go
View File

@ -255,6 +255,7 @@ type Group struct {
Ports []int `mapstructure:"ports"`
} `mapstructure:"rpc"`
Prometheus Prometheus `mapstructure:"prometheus"`
GroupSyncCount int `mapstructure:"groupSyncCount"`
}
type Msg struct {

3
pkg/common/storage/cache/group.go vendored
View File

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

23
pkg/common/storage/cache/redis/group.go vendored
View File

@ -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
}

41
pkg/common/storage/controller/group.go
View File

@ -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 {
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
}
if deleteMember {
userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID)
if err != nil {
return err
}
if deleteMember {
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)
}

3
pkg/common/storage/database/group_member.go
View File

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

10
pkg/common/storage/database/mgo/friend.go
View File

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

85
pkg/common/storage/database/mgo/group_member.go
View File

@ -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
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.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.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 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) {
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 (g *GroupMemberMgo) Find(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32) (groupMembers []*model.GroupMember, err error) {
// TODO implement me
panic("implement me")
}, 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)
}

2
pkg/common/storage/database/mgo/version_log.go
View File

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

2
pkg/common/storage/database/version_log.go
View File

@ -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
}