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

group version

Browse Source
This commit is contained in:
withchao 2024-06-17 15:28:09 +08:00
parent 1b1027192d
commit a41c8c6a9d
10 changed files with 49 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
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.9
github.com/openimsdk/protocol v0.0.69-alpha.12
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

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.9 h1:Mh1upsxwhWs1y65cfIN2XuRsiKEk4MfpYfA9XrvtX24=
github.com/openimsdk/protocol v0.0.69-alpha.9/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
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/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=

8
internal/api/group.go
View File

@ -187,10 +187,10 @@ func (o *GroupApi) GetIncrementalGroupMemberBatch(c *gin.Context) {
apiresp.GinSuccess(c, resp)
}
func (o *GroupApi) GetIncrementalGroupMemberUserIDs(c *gin.Context) {
a2r.Call(group.GroupClient.GetIncrementalGroupMemberUserIDs, o.Client, c)
func (o *GroupApi) GetFullGroupMemberUserIDs(c *gin.Context) {
a2r.Call(group.GroupClient.GetFullGroupMemberUserIDs, o.Client, c)
}
func (o *GroupApi) GetIncrementalJoinGroupIDs(c *gin.Context) {
a2r.Call(group.GroupClient.GetIncrementalJoinGroupIDs, o.Client, c)
func (o *GroupApi) GetFullJoinGroupIDs(c *gin.Context) {
a2r.Call(group.GroupClient.GetFullJoinGroupIDs, o.Client, c)
}

4
internal/api/router.go
View File

@ -120,8 +120,8 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
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)
groupRouterGroup.POST("/get_incremental_group_member_user_ids", g.GetIncrementalGroupMemberUserIDs)
groupRouterGroup.POST("/get_incremental_join_group_ids", g.GetIncrementalJoinGroupIDs)
groupRouterGroup.POST("/get_incremental_group_member_user_ids", g.GetFullGroupMemberUserIDs)
groupRouterGroup.POST("/get_incremental_join_group_ids", g.GetFullJoinGroupIDs)
}
// certificate
authRouterGroup := r.Group("/auth")

5
internal/rpc/friend/friend.go
View File

@ -51,6 +51,11 @@ type friendServer struct {
webhookClient *webhook.Client
}
func (s *friendServer) GetFullFriendUserIDs(ctx context.Context, req *relation.GetFullFriendUserIDsReq) (*relation.GetFullFriendUserIDsResp, error) {
//TODO implement me
panic("implement me")
}
type Config struct {
RpcConfig config.Friend
RedisConfig config.Redis

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

@ -289,10 +289,12 @@ func (g *GroupNotificationSender) fillOpUser(ctx context.Context, opUser **sdkws
return nil
}
func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint64, collName string, id string) {
for _, coll := range versionctx.GetVersionLog(ctx).Get() {
func (g *GroupNotificationSender) setVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string) {
versions := versionctx.GetVersionLog(ctx).Get()
for _, coll := range versions {
if coll.Name == collName && coll.Doc.DID == id {
*version = uint64(coll.Doc.Version)
*versionID = coll.Doc.ID.Hex()
return
}
}
@ -308,7 +310,7 @@ func (g *GroupNotificationSender) GroupCreatedNotification(ctx context.Context,
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupCreatedNotification, tips)
}
@ -392,7 +394,7 @@ func (g *GroupNotificationSender) MemberQuitNotification(ctx context.Context, me
return
}
tips := &sdkws.MemberQuitTips{Group: group, QuitUser: member}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, member.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, member.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), member.GroupID, constant.MemberQuitNotification, tips)
}
@ -480,7 +482,7 @@ func (g *GroupNotificationSender) GroupOwnerTransferredNotification(ctx context.
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, req.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, req.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupOwnerTransferredNotification, tips)
}
@ -494,7 +496,7 @@ func (g *GroupNotificationSender) MemberKickedNotification(ctx context.Context,
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.MemberKickedNotification, tips)
}
@ -518,7 +520,7 @@ func (g *GroupNotificationSender) MemberInvitedNotification(ctx context.Context,
}
tips := &sdkws.MemberInvitedTips{Group: group, InvitedUserList: users}
err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberInvitedNotification, tips)
}
@ -540,7 +542,7 @@ func (g *GroupNotificationSender) MemberEnterNotification(ctx context.Context, g
return
}
tips := &sdkws.MemberEnterTips{Group: group, EntrantUser: user}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.MemberEnterNotification, tips)
}
@ -581,7 +583,7 @@ func (g *GroupNotificationSender) GroupMemberMutedNotification(ctx context.Conte
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberMutedNotification, tips)
}
@ -606,7 +608,7 @@ func (g *GroupNotificationSender) GroupMemberCancelMutedNotification(ctx context
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberCancelMutedNotification, tips)
}
@ -685,7 +687,7 @@ func (g *GroupNotificationSender) GroupMemberInfoSetNotification(ctx context.Con
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberInfoSetNotification, tips)
}
@ -709,7 +711,7 @@ func (g *GroupNotificationSender) GroupMemberSetToAdminNotification(ctx context.
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToAdminNotification, tips)
}
@ -734,6 +736,6 @@ func (g *GroupNotificationSender) GroupMemberSetToOrdinaryUserNotification(ctx c
if err = g.fillOpUser(ctx, &tips.OpUser, tips.Group.GroupID); err != nil {
return
}
g.setVersion(ctx, &tips.GroupMemberVersion, database.GroupMemberVersionName, tips.Group.GroupID)
g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID)
g.Notification(ctx, mcontext.GetOpUserID(ctx), group.GroupID, constant.GroupMemberSetToOrdinaryUserNotification, tips)
}

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

@ -21,7 +21,7 @@ func (s *groupServer) idHash(ids []string) uint64 {
return binary.BigEndian.Uint64(sum[:])
}
func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetIncrementalGroupMemberUserIDsReq) (*pbgroup.GetIncrementalGroupMemberUserIDsResp, error) {
func (s *groupServer) GetFullGroupMemberUserIDs(ctx context.Context, req *pbgroup.GetFullGroupMemberUserIDsReq) (*pbgroup.GetFullGroupMemberUserIDsResp, error) {
vl, err := s.db.FindMaxGroupMemberVersionCache(ctx, req.GroupID)
if err != nil {
return nil, err
@ -34,7 +34,7 @@ func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req
if req.IdHash == idHash {
userIDs = nil
}
return &pbgroup.GetIncrementalGroupMemberUserIDsResp{
return &pbgroup.GetFullGroupMemberUserIDsResp{
Version: idHash,
VersionID: vl.ID.Hex(),
Equal: req.IdHash == idHash,
@ -42,7 +42,7 @@ func (s *groupServer) GetIncrementalGroupMemberUserIDs(ctx context.Context, req
}, nil
}
func (s *groupServer) GetIncrementalJoinGroupIDs(ctx context.Context, req *pbgroup.GetIncrementalJoinGroupIDsReq) (*pbgroup.GetIncrementalJoinGroupIDsResp, error) {
func (s *groupServer) GetFullJoinGroupIDs(ctx context.Context, req *pbgroup.GetFullJoinGroupIDsReq) (*pbgroup.GetFullJoinGroupIDsResp, error) {
vl, err := s.db.FindMaxJoinGroupVersionCache(ctx, req.UserID)
if err != nil {
return nil, err
@ -55,7 +55,7 @@ func (s *groupServer) GetIncrementalJoinGroupIDs(ctx context.Context, req *pbgro
if req.IdHash == idHash {
groupIDs = nil
}
return &pbgroup.GetIncrementalJoinGroupIDsResp{
return &pbgroup.GetFullJoinGroupIDsResp{
Version: idHash,
VersionID: vl.ID.Hex(),
Equal: req.IdHash == idHash,

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

@ -96,13 +96,19 @@ func (g *GroupMemberMgo) Delete(ctx context.Context, groupID string, userIDs []s
}
return mongoutil.IncrVersion(func() error {
return mongoutil.DeleteMany(ctx, g.coll, filter)
}, func() error {
return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete)
}, func() error {
if len(userIDs) == 0 {
return nil
return g.member.Delete(ctx, groupID)
} else {
return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete)
}
return g.member.IncrVersion(ctx, groupID, userIDs, model.VersionStateDelete)
}, func() error {
for _, userID := range userIDs {
if err := g.join.IncrVersion(ctx, userID, []string{groupID}, model.VersionStateDelete); err != nil {
return err
}
}
return nil
})
}
@ -112,7 +118,7 @@ func (g *GroupMemberMgo) UpdateRoleLevel(ctx context.Context, groupID string, us
}, func() error {
return g.member.IncrVersion(ctx, groupID, []string{userID}, model.VersionStateUpdate)
}, func() error {
return g.join.IncrVersion(ctx, groupID, []string{userID}, model.VersionStateUpdate)
return g.join.IncrVersion(ctx, userID, []string{groupID}, model.VersionStateUpdate)
})
}

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

@ -256,3 +256,7 @@ func (l *VersionLogMgo) DeleteAfterUnchangedLog(ctx context.Context, deadline ti
},
})
}
func (l *VersionLogMgo) Delete(ctx context.Context, dId string) error {
return mongoutil.DeleteOne(ctx, l.coll, bson.M{"d_id": dId})
}

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

@ -15,4 +15,5 @@ type VersionLog interface {
IncrVersion(ctx context.Context, dId string, eIds []string, state int32) error
FindChangeLog(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error)
DeleteAfterUnchangedLog(ctx context.Context, deadline time.Time) error
Delete(ctx context.Context, dId string) error
}