blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-11-04 11:22:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

group

Browse Source
This commit is contained in:
withchao 2024-06-19 16:28:28 +08:00
parent aff8322bc5
commit 08cb4a948d
9 changed files with 148 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
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.12
github.com/openimsdk/protocol v0.0.69-alpha.15
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
@ -176,3 +176,7 @@ require (
golang.org/x/crypto v0.21.0 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
)
//replace (
// github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol
//)

4
go.sum
View File

@ -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.12 h1:3ZdwmD1y9vcduIC8o2EZS8Ds/fByqcuEFo+NkcBzgRo=
github.com/openimsdk/protocol v0.0.69-alpha.12/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
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/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=

5
internal/rpc/group/notification.go
View File

@ -324,6 +324,7 @@ func (g *GroupNotificationSender) GroupInfoSetNotification(ctx context.Context,
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetNotification, tips, rpcclient.WithRpcGetUserName())
}
@ -337,6 +338,7 @@ func (g *GroupNotificationSender) GroupInfoSetNameNotification(ctx context.Conte
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetNameNotification, tips)
}
@ -350,6 +352,7 @@ func (g *GroupNotificationSender) GroupInfoSetAnnouncementNotification(ctx conte
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetAnnouncementNotification, tips, rpcclient.WithRpcGetUserName())
}
@ -636,6 +639,7 @@ func (g *GroupNotificationSender) GroupMutedNotification(ctx context.Context, gr
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, groupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMutedNotification, tips)
}
@ -663,6 +667,7 @@ func (g *GroupNotificationSender) GroupCancelMutedNotification(ctx context.Conte
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, groupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupCancelMutedNotification, tips)
}

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

@ -4,10 +4,13 @@ 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/servererrs"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"github.com/openimsdk/open-im-server/v3/pkg/util/hashutil"
"github.com/openimsdk/protocol/constant"
pbgroup "github.com/openimsdk/protocol/group"
"github.com/openimsdk/protocol/sdkws"
"slices"
)
func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) {
@ -53,12 +56,36 @@ func (s *groupServer) GetFullJoinGroupIDs(ctx context.Context, req *pbgroup.GetF
}
func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgroup.GetIncrementalGroupMemberReq) (*pbgroup.GetIncrementalGroupMemberResp, error) {
group, err := s.db.TakeGroup(ctx, req.GroupID)
if err != nil {
return nil, err
}
if group.Status == constant.GroupStatusDismissed {
return nil, servererrs.ErrDismissedAlready.Wrap()
}
opt := incrversion.Option[*sdkws.GroupMemberFullInfo, pbgroup.GetIncrementalGroupMemberResp]{
Ctx: ctx,
VersionKey: req.GroupID,
VersionID: req.VersionID,
VersionNumber: req.Version,
Version: s.db.FindMemberIncrVersion,
Ctx: ctx,
VersionKey: req.GroupID,
VersionID: req.VersionID,
VersionNumber: req.Version,
Version: func(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) {
vl, err := s.db.FindMemberIncrVersion(ctx, groupID, version, limit)
if err != nil {
return nil, err
}
var ok bool
vl.Logs = slices.DeleteFunc(vl.Logs, func(elem model.VersionLogElem) bool {
if elem.EID == "" {
ok = true
return true
}
return false
})
if !ok {
group = nil
}
return vl, nil
},
CacheMaxVersion: s.db.FindMaxGroupMemberVersionCache,
Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) {
return s.getGroupMembersInfo(ctx, req.GroupID, ids)
@ -75,7 +102,22 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
}
},
}
return opt.Build()
resp, err := opt.Build()
if err != nil {
return nil, err
}
if group != nil {
count, err := s.db.FindGroupMemberNum(ctx, group.GroupID)
if err != nil {
return nil, err
}
owner, err := s.db.TakeGroupOwner(ctx, group.GroupID)
if err != nil {
return nil, err
}
resp.Group = s.groupDB2PB(group, owner.UserID, count)
}
return resp, nil
}
func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupReq) (*pbgroup.GetIncrementalJoinGroupResp, error) {

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

@ -243,17 +243,10 @@ func (g *groupDatabase) UpdateGroup(ctx context.Context, groupID string, data ma
if err := g.groupDB.UpdateMap(ctx, groupID, data); err != nil {
return err
}
userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID)
if err != nil {
if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, []string{""}, model.VersionStateUpdate); err != nil {
return err
}
for _, userID := range userIDs {
if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, userID, []string{groupID}, model.VersionStateUpdate); err != nil {
return err
}
}
return g.cache.CloneGroupCache().DelGroupsInfo(groupID).DelMaxJoinGroupVersion(userIDs...).ChainExecDel(ctx)
return g.cache.CloneGroupCache().DelGroupsInfo(groupID).DelMaxGroupMemberVersion(groupID).ChainExecDel(ctx)
})
}
@ -263,11 +256,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
}
@ -277,13 +270,18 @@ func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, delete
DelGroupMembersHash(groupID).
DelGroupAllRoleLevel(groupID).
DelGroupMembersInfo(groupID, userIDs...).
DelMaxGroupMemberVersion(groupID)
}
c = c.DelMaxJoinGroupVersion(userIDs...)
if len(userIDs) > 0 {
if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, groupID, userIDs, model.VersionStateDelete); err != nil {
DelMaxGroupMemberVersion(groupID).
DelMaxJoinGroupVersion(userIDs...)
for _, userID := range userIDs {
if err := g.groupMemberDB.JoinGroupIncrVersion(ctx, userID, []string{groupID}, model.VersionStateDelete); err != nil {
return err
}
}
} else {
if err := g.groupMemberDB.MemberGroupIncrVersion(ctx, groupID, []string{""}, model.VersionStateUpdate); err != nil {
return err
}
c = c.DelMaxGroupMemberVersion(groupID)
}
return c.DelGroupsInfo(groupID).ChainExecDel(ctx)
})
@ -411,12 +409,16 @@ func (g *groupDatabase) TransferGroupOwner(ctx context.Context, groupID string,
return c.DelGroupMembersInfo(groupID, oldOwnerUserID, newOwnerUserID).
DelGroupAllRoleLevel(groupID).
DelGroupMembersHash(groupID).
DelJoinedGroupID(oldOwnerUserID, newOwnerUserID).
DelMaxGroupMemberVersion(groupID).
DelGroupMemberIDs(groupID).
ChainExecDel(ctx)
})
}
func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, userID string, data map[string]any) error {
if len(data) == 0 {
return nil
}
return g.ctxTx.Transaction(ctx, func(ctx context.Context) error {
if err := g.groupMemberDB.Update(ctx, groupID, userID, data); err != nil {
return err
@ -424,9 +426,9 @@ func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, u
c := g.cache.CloneGroupCache()
c = c.DelGroupMembersInfo(groupID, userID)
if g.groupMemberDB.IsUpdateRoleLevel(data) {
c = c.DelGroupAllRoleLevel(groupID)
c = c.DelGroupAllRoleLevel(groupID).DelGroupMemberIDs(groupID)
}
c = c.DelMaxGroupMemberVersion(groupID).DelMaxJoinGroupVersion(userID)
c = c.DelMaxGroupMemberVersion(groupID)
return c.ChainExecDel(ctx)
})
}
@ -439,9 +441,9 @@ func (g *groupDatabase) UpdateGroupMembers(ctx context.Context, data []*common.B
return err
}
if g.groupMemberDB.IsUpdateRoleLevel(item.Map) {
c = c.DelGroupAllRoleLevel(item.GroupID)
c = c.DelGroupAllRoleLevel(item.GroupID).DelGroupMemberIDs(item.GroupID)
}
c = c.DelGroupMembersInfo(item.GroupID, item.UserID).DelGroupMembersHash(item.GroupID)
c = c.DelGroupMembersInfo(item.GroupID, item.UserID).DelMaxGroupMemberVersion(item.GroupID).DelGroupMembersHash(item.GroupID)
}
return c.ChainExecDel(ctx)
})
@ -501,14 +503,6 @@ func (g *groupDatabase) FindJoinIncrVersion(ctx context.Context, userID string,
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)
//}
func (g *groupDatabase) FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error) {
return g.cache.FindMaxGroupMemberVersion(ctx, groupID)
}

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

@ -35,6 +35,7 @@ type GroupMember interface {
FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error)
IsUpdateRoleLevel(data map[string]any) bool
JoinGroupIncrVersion(ctx context.Context, userID string, groupIDs []string, state int32) error
MemberGroupIncrVersion(ctx context.Context, groupID string, userIDs []string, state int32) 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)
}

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

@ -18,6 +18,8 @@ import (
"context"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
"go.mongodb.org/mongo-driver/bson/primitive"
"time"
"github.com/openimsdk/tools/db/mongoutil"
"github.com/openimsdk/tools/db/pagination"
@ -53,11 +55,19 @@ func NewFriendMongo(db *mongo.Database) (database.Friend, error) {
}
func (f *FriendMgo) friendSort() any {
return bson.D{{"is_pinned", -1}, {"create_time", 1}, {"_id", 1}}
return bson.D{{"is_pinned", -1}, {"_id", 1}}
}
// Create inserts multiple friend records.
func (f *FriendMgo) Create(ctx context.Context, friends []*model.Friend) error {
for i, friend := range friends {
if friend.ID.IsZero() {
friends[i].ID = primitive.NewObjectID()
}
if friend.CreateTime.IsZero() {
friends[i].CreateTime = time.Now()
}
}
return mongoutil.IncrVersion(func() error {
return mongoutil.InsertMany(ctx, f.coll, friends)
}, func() error {
@ -108,13 +118,43 @@ func (f *FriendMgo) UpdateRemark(ctx context.Context, ownerUserID, friendUserID,
return f.UpdateByMap(ctx, ownerUserID, friendUserID, map[string]any{"remark": remark})
}
func (f *FriendMgo) fillTime(friends ...*model.Friend) {
for i, friend := range friends {
if friend.CreateTime.IsZero() {
friends[i].CreateTime = friend.ID.Timestamp()
}
}
}
func (f *FriendMgo) findOne(ctx context.Context, filter any) (*model.Friend, error) {
friend, err := mongoutil.FindOne[*model.Friend](ctx, f.coll, filter)
if err != nil {
return nil, err
}
f.fillTime(friend)
return friend, nil
}
func (f *FriendMgo) find(ctx context.Context, filter any) ([]*model.Friend, error) {
friends, err := mongoutil.Find[*model.Friend](ctx, f.coll, filter)
if err != nil {
return nil, err
}
f.fillTime(friends...)
return friends, nil
}
func (f *FriendMgo) findPage(ctx context.Context, filter any, pagination pagination.Pagination, opts ...*options.FindOptions) (int64, []*model.Friend, error) {
return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination, opts...)
}
// Take retrieves a single friend document. Returns an error if not found.
func (f *FriendMgo) Take(ctx context.Context, ownerUserID, friendUserID string) (*model.Friend, error) {
filter := bson.M{
"owner_user_id": ownerUserID,
"friend_user_id": friendUserID,
}
return mongoutil.FindOne[*model.Friend](ctx, f.coll, filter)
return f.findOne(ctx, filter)
}
// FindUserState finds the friendship status between two users.
@ -125,7 +165,7 @@ func (f *FriendMgo) FindUserState(ctx context.Context, userID1, userID2 string)
{"owner_user_id": userID2, "friend_user_id": userID1},
},
}
return mongoutil.Find[*model.Friend](ctx, f.coll, filter)
return f.find(ctx, filter)
}
// FindFriends retrieves a list of friends for a given owner. Missing friends do not cause an error.
@ -134,7 +174,7 @@ func (f *FriendMgo) FindFriends(ctx context.Context, ownerUserID string, friendU
"owner_user_id": ownerUserID,
"friend_user_id": bson.M{"$in": friendUserIDs},
}
return mongoutil.Find[*model.Friend](ctx, f.coll, filter)
return f.find(ctx, filter)
}
// FindReversalFriends finds users who have added the specified user as a friend.
@ -143,14 +183,14 @@ func (f *FriendMgo) FindReversalFriends(ctx context.Context, friendUserID string
"owner_user_id": bson.M{"$in": ownerUserIDs},
"friend_user_id": friendUserID,
}
return mongoutil.Find[*model.Friend](ctx, f.coll, filter)
return f.find(ctx, filter)
}
// FindOwnerFriends retrieves a paginated list of friends for a given owner.
func (f *FriendMgo) FindOwnerFriends(ctx context.Context, ownerUserID string, pagination pagination.Pagination) (int64, []*model.Friend, error) {
filter := bson.M{"owner_user_id": ownerUserID}
opt := options.Find().SetSort(f.friendSort())
return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination, opt)
return f.findPage(ctx, filter, pagination, opt)
}
func (f *FriendMgo) FindOwnerFriendUserIds(ctx context.Context, ownerUserID string, limit int) ([]string, error) {
@ -162,7 +202,8 @@ func (f *FriendMgo) FindOwnerFriendUserIds(ctx context.Context, ownerUserID stri
// FindInWhoseFriends finds users who have added the specified user as a friend, with pagination.
func (f *FriendMgo) FindInWhoseFriends(ctx context.Context, friendUserID string, pagination pagination.Pagination) (int64, []*model.Friend, error) {
filter := bson.M{"friend_user_id": friendUserID}
return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination)
opt := options.Find().SetSort(f.friendSort())
return f.findPage(ctx, filter, pagination, opt)
}
// FindFriendUserIDs retrieves a list of friend user IDs for a given owner.
@ -203,18 +244,3 @@ 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) SearchFriend(ctx context.Context, ownerUserID, keyword string, pagination pagination.Pagination) (int64, []*model.Friend, error) {
// filter := bson.M{
// "owner_user_id": ownerUserID,
// }
// if keyword != "" {
// filter["$or"] = []bson.M{
// {"remark": bson.M{"$regex": keyword, "$options": "i"}},
// {"nickname": bson.M{"$regex": keyword, "$options": "i"}},
// {"friend_user_id": bson.M{"$regex": keyword, "$options": "i"}},
// }
// }
// opt := options.Find().SetSort(f.friendSort())
// return mongoutil.FindPage[*model.Friend](ctx, f.coll, filter, pagination, opt)
//}

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

@ -117,8 +117,6 @@ func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, us
return g.Update(ctx, groupID, userID, bson.M{"role_level": roleLevel})
}, func() error {
return g.member.IncrVersion(ctx, groupID, []string{userID}, model.VersionStateUpdate)
}, func() error {
return g.join.IncrVersion(ctx, userID, []string{groupID}, model.VersionStateUpdate)
})
}
@ -184,6 +182,10 @@ func (g *GroupMemberMgo) JoinGroupIncrVersion(ctx context.Context, userID string
return g.join.IncrVersion(ctx, userID, groupIDs, state)
}
func (g *GroupMemberMgo) MemberGroupIncrVersion(ctx context.Context, groupID string, userIDs []string, state int32) error {
return g.member.IncrVersion(ctx, groupID, userIDs, state)
}
func (g *GroupMemberMgo) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) {
return g.member.FindChangeLog(ctx, groupID, version, limit)
}

18
pkg/common/storage/model/friend.go
View File

@ -15,17 +15,19 @@
package model
import (
"go.mongodb.org/mongo-driver/bson/primitive"
"time"
)
// Friend represents the data structure for a friend relationship in MongoDB.
type Friend struct {
OwnerUserID string `bson:"owner_user_id"`
FriendUserID string `bson:"friend_user_id"`
Remark string `bson:"remark"`
CreateTime time.Time `bson:"create_time"`
AddSource int32 `bson:"add_source"`
OperatorUserID string `bson:"operator_user_id"`
Ex string `bson:"ex"`
IsPinned bool `bson:"is_pinned"`
ID primitive.ObjectID `bson:"_id"`
OwnerUserID string `bson:"owner_user_id"`
FriendUserID string `bson:"friend_user_id"`
Remark string `bson:"remark"`
CreateTime time.Time `bson:"create_time"`
AddSource int32 `bson:"add_source"`
OperatorUserID string `bson:"operator_user_id"`
Ex string `bson:"ex"`
IsPinned bool `bson:"is_pinned"`
}