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

GROUP

Browse Source
This commit is contained in:
wangchuxiao 2023-03-27 11:03:50 +08:00
parent 96321a2469
commit a999b9d0c4
2 changed files with 35 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
internal/tools/msg.go
View File

@ -4,6 +4,8 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"math"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
@ -14,7 +16,6 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils" "github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"github.com/go-redis/redis/v8" "github.com/go-redis/redis/v8"
"math"
) )
type MsgTool struct { type MsgTool struct {
@ -53,35 +54,30 @@ func InitMsgTool() (*MsgTool, error) {
return msgTool, nil return msgTool, nil
} }
func (c *MsgTool) getCronTaskOperationID() string {
return cronTaskOperationID + utils.OperationIDGenerator()
}
func (c *MsgTool) AllUserClearMsgAndFixSeq() { func (c *MsgTool) AllUserClearMsgAndFixSeq() {
operationID := "AllUserAndGroupClearMsgAndFixSeq"
ctx := mcontext.NewCtx(utils.GetSelfFuncName()) ctx := mcontext.NewCtx(utils.GetSelfFuncName())
log.NewInfo(operationID, "============================ start del cron task ============================") log.ZInfo(ctx, "============================ start del cron task ============================")
var err error var err error
userIDList, err := c.userDatabase.GetAllUserID(ctx) userIDs, err := c.userDatabase.GetAllUserID(ctx)
if err == nil { if err == nil {
c.ClearUsersMsg(ctx, userIDList) c.ClearUsersMsg(ctx, userIDs)
} else { } else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error()) log.ZError(ctx, "ClearUsersMsg failed", err)
} }
// working group msg clear // working group msg clear
superGroupIDList, err := c.groupDatabase.GetGroupIDsByGroupType(ctx, constant.WorkingGroup) superGroupIDs, err := c.groupDatabase.GetGroupIDsByGroupType(ctx, constant.WorkingGroup)
if err == nil { if err == nil {
c.ClearSuperGroupMsg(ctx, superGroupIDList) c.ClearSuperGroupMsg(ctx, superGroupIDs)
} else { } else {
log.NewError(operationID, utils.GetSelfFuncName(), err.Error()) log.ZError(ctx, "ClearSuperGroupMsg failed", err)
} }
log.NewInfo(operationID, "============================ start del cron finished ============================") log.ZInfo(ctx, "============================ start del cron finished ============================")
} }
func (c *MsgTool) ClearUsersMsg(ctx context.Context, userIDs []string) { func (c *MsgTool) ClearUsersMsg(ctx context.Context, userIDs []string) {
for _, userID := range userIDs { for _, userID := range userIDs {
if err := c.msgDatabase.DeleteUserMsgsAndSetMinSeq(ctx, userID, int64(config.Config.Mongo.DBRetainChatRecords*24*60*60)); err != nil { if err := c.msgDatabase.DeleteUserMsgsAndSetMinSeq(ctx, userID, int64(config.Config.Mongo.DBRetainChatRecords*24*60*60)); err != nil {
log.NewError(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), userID) log.ZError(ctx, "DeleteUserMsgsAndSetMinSeq failed", err, "userID", userID, "DBRetainChatRecords", config.Config.Mongo.DBRetainChatRecords)
} }
maxSeqCache, maxSeqMongo, err := c.GetAndFixUserSeqs(ctx, userID) maxSeqCache, maxSeqMongo, err := c.GetAndFixUserSeqs(ctx, userID)
if err != nil { if err != nil {
@ -95,14 +91,14 @@ func (c *MsgTool) ClearSuperGroupMsg(ctx context.Context, superGroupIDs []string
for _, groupID := range superGroupIDs { for _, groupID := range superGroupIDs {
userIDs, err := c.groupDatabase.FindGroupMemberUserID(ctx, groupID) userIDs, err := c.groupDatabase.FindGroupMemberUserID(ctx, groupID)
if err != nil { if err != nil {
log.NewError(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), "FindGroupMemberUserID", err.Error(), groupID) log.ZError(ctx, "ClearSuperGroupMsg failed", err, "groupID", groupID)
continue continue
} }
if err := c.msgDatabase.DeleteUserSuperGroupMsgsAndSetMinSeq(ctx, groupID, userIDs, int64(config.Config.Mongo.DBRetainChatRecords*24*60*60)); err != nil { if err := c.msgDatabase.DeleteUserSuperGroupMsgsAndSetMinSeq(ctx, groupID, userIDs, int64(config.Config.Mongo.DBRetainChatRecords*24*60*60)); err != nil {
log.NewError(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), "DeleteUserSuperGroupMsgsAndSetMinSeq failed", groupID, userIDs, config.Config.Mongo.DBRetainChatRecords) log.ZError(ctx, "DeleteUserSuperGroupMsgsAndSetMinSeq failed", err, "groupID", groupID, "userID", userIDs, "DBRetainChatRecords", config.Config.Mongo.DBRetainChatRecords)
} }
if err := c.fixGroupSeq(ctx, groupID, userIDs); err != nil { if err := c.fixGroupSeq(ctx, groupID, userIDs); err != nil {
log.NewError(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), groupID, userIDs) log.ZError(ctx, "fixGroupSeq failed", err, "groupID", groupID, "userID", userIDs)
} }
} }
} }
@ -129,7 +125,7 @@ func (c *MsgTool) fixGroupSeq(ctx context.Context, groupID string, userIDs []str
} }
} }
if err := c.CheckMaxSeqWithMongo(ctx, groupID, maxSeqCache, maxSeqMongo, constant.WriteDiffusion); err != nil { if err := c.CheckMaxSeqWithMongo(ctx, groupID, maxSeqCache, maxSeqMongo, constant.WriteDiffusion); err != nil {
log.NewWarn(mcontext.GetOperationID(ctx), "cache max seq and mongo max seq is diff > 10", groupID, maxSeqCache, maxSeqMongo, constant.WriteDiffusion) log.ZWarn(ctx, "cache max seq and mongo max seq is diff > 10", err, "groupID", groupID, "maxSeqCache", maxSeqCache, "maxSeqMongo", maxSeqMongo, "constant.WriteDiffusion", constant.WriteDiffusion)
} }
return nil return nil
} }
@ -138,16 +134,16 @@ func (c *MsgTool) GetAndFixUserSeqs(ctx context.Context, userID string) (maxSeqC
minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache, err := c.msgDatabase.GetUserMinMaxSeqInMongoAndCache(ctx, userID) minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache, err := c.msgDatabase.GetUserMinMaxSeqInMongoAndCache(ctx, userID)
if err != nil { if err != nil {
if err != unrelation.ErrMsgNotFound { if err != unrelation.ErrMsgNotFound {
log.NewError(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), "GetUserMinMaxSeqInMongoAndCache failed", userID) log.ZError(ctx, "GetUserMinMaxSeqInMongoAndCache failed", err, "userID", userID)
} }
return 0, 0, err return 0, 0, err
} }
log.NewDebug(mcontext.GetOperationID(ctx), userID, minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache) log.ZDebug(ctx, "userID", userID, "minSeqMongo", minSeqMongo, "maxSeqMongo", maxSeqMongo, "minSeqCache", minSeqCache, "maxSeqCache", maxSeqCache)
if minSeqCache > maxSeqCache { if minSeqCache > maxSeqCache {
if err := c.msgDatabase.SetUserMinSeq(ctx, userID, maxSeqCache); err != nil { if err := c.msgDatabase.SetUserMinSeq(ctx, userID, maxSeqCache); err != nil {
log.NewError(mcontext.GetOperationID(ctx), "SetUserMinSeq failed", userID, minSeqCache, maxSeqCache) log.ZError(ctx, "SetUserMinSeq failed", err, "userID", userID, "minSeqCache", minSeqCache, "maxSeqCache", maxSeqCache)
} else { } else {
log.NewWarn(mcontext.GetOperationID(ctx), "SetUserMinSeq success", userID, minSeqCache, maxSeqCache) log.ZInfo(ctx, "SetUserMinSeq success", "userID", userID, "minSeqCache", minSeqCache, "maxSeqCache", maxSeqCache)
} }
} }
return maxSeqCache, maxSeqMongo, nil return maxSeqCache, maxSeqMongo, nil
@ -156,14 +152,14 @@ func (c *MsgTool) GetAndFixUserSeqs(ctx context.Context, userID string) (maxSeqC
func (c *MsgTool) GetAndFixGroupUserSeq(ctx context.Context, userID string, groupID string, maxSeqCache int64) (minSeqCache int64, err error) { func (c *MsgTool) GetAndFixGroupUserSeq(ctx context.Context, userID string, groupID string, maxSeqCache int64) (minSeqCache int64, err error) {
minSeqCache, err = c.msgDatabase.GetGroupUserMinSeq(ctx, groupID, userID) minSeqCache, err = c.msgDatabase.GetGroupUserMinSeq(ctx, groupID, userID)
if err != nil { if err != nil {
log.NewError(mcontext.GetOperationID(ctx), "GetGroupUserMinSeq failed", groupID, userID) log.ZError(ctx, "GetGroupUserMinSeq failed", err, "groupID", groupID, "userID", userID)
return 0, err return 0, err
} }
if minSeqCache > maxSeqCache { if minSeqCache > maxSeqCache {
if err := c.msgDatabase.SetGroupUserMinSeq(ctx, groupID, userID, maxSeqCache); err != nil { if err := c.msgDatabase.SetGroupUserMinSeq(ctx, groupID, userID, maxSeqCache); err != nil {
log.NewError(mcontext.GetOperationID(ctx), "SetGroupUserMinSeq failed", userID, minSeqCache, maxSeqCache) log.ZError(ctx, "SetGroupUserMinSeq failed", err, "groupID", groupID, "userID", userID, "minSeqCache", minSeqCache, "maxSeqCache", maxSeqCache)
} else { } else {
log.NewWarn(mcontext.GetOperationID(ctx), "SetGroupUserMinSeq success", userID, minSeqCache, maxSeqCache) log.ZInfo(ctx, "SetGroupUserMinSeq success", "groupID", groupID, "userID", userID, "minSeqCache", minSeqCache, "maxSeqCache", maxSeqCache)
} }
} }
return minSeqCache, nil return minSeqCache, nil

26
pkg/common/db/relation/group_member_model.go
View File

@ -25,23 +25,23 @@ func (g *GroupMemberGorm) NewTx(tx any) relation.GroupMemberModelInterface {
} }
func (g *GroupMemberGorm) Create(ctx context.Context, groupMemberList []*relation.GroupMemberModel) (err error) { func (g *GroupMemberGorm) Create(ctx context.Context, groupMemberList []*relation.GroupMemberModel) (err error) {
return utils.Wrap(g.DB.Create(&groupMemberList).Error, "") return utils.Wrap(g.db(ctx).Create(&groupMemberList).Error, "")
} }
func (g *GroupMemberGorm) Delete(ctx context.Context, groupID string, userIDs []string) (err error) { func (g *GroupMemberGorm) Delete(ctx context.Context, groupID string, userIDs []string) (err error) {
return utils.Wrap(g.DB.Where("group_id = ? and user_id in (?)", groupID, userIDs).Delete(&relation.GroupMemberModel{}).Error, "") return utils.Wrap(g.db(ctx).Where("group_id = ? and user_id in (?)", groupID, userIDs).Delete(&relation.GroupMemberModel{}).Error, "")
} }
func (g *GroupMemberGorm) DeleteGroup(ctx context.Context, groupIDs []string) (err error) { func (g *GroupMemberGorm) DeleteGroup(ctx context.Context, groupIDs []string) (err error) {
return utils.Wrap(g.DB.Where("group_id in (?)", groupIDs).Delete(&relation.GroupMemberModel{}).Error, "") return utils.Wrap(g.db(ctx).Where("group_id in (?)", groupIDs).Delete(&relation.GroupMemberModel{}).Error, "")
} }
func (g *GroupMemberGorm) Update(ctx context.Context, groupID string, userID string, data map[string]any) (err error) { func (g *GroupMemberGorm) Update(ctx context.Context, groupID string, userID string, data map[string]any) (err error) {
return utils.Wrap(g.DB.Model(&relation.GroupMemberModel{}).Where("group_id = ? and user_id = ?", groupID, userID).Updates(data).Error, "") return utils.Wrap(g.db(ctx).Where("group_id = ? and user_id = ?", groupID, userID).Updates(data).Error, "")
} }
func (g *GroupMemberGorm) UpdateRoleLevel(ctx context.Context, groupID string, userID string, roleLevel int32) (rowsAffected int64, err error) { func (g *GroupMemberGorm) UpdateRoleLevel(ctx context.Context, groupID string, userID string, roleLevel int32) (rowsAffected int64, err error) {
db := g.DB.Model(&relation.GroupMemberModel{}).Where("group_id = ? and user_id = ?", groupID, userID).Updates(map[string]any{ db := g.db(ctx).Where("group_id = ? and user_id = ?", groupID, userID).Updates(map[string]any{
"role_level": roleLevel, "role_level": roleLevel,
}) })
return db.RowsAffected, utils.Wrap(db.Error, "") return db.RowsAffected, utils.Wrap(db.Error, "")
@ -63,16 +63,16 @@ func (g *GroupMemberGorm) Find(ctx context.Context, groupIDs []string, userIDs [
func (g *GroupMemberGorm) Take(ctx context.Context, groupID string, userID string) (groupMember *relation.GroupMemberModel, err error) { func (g *GroupMemberGorm) Take(ctx context.Context, groupID string, userID string) (groupMember *relation.GroupMemberModel, err error) {
groupMember = &relation.GroupMemberModel{} groupMember = &relation.GroupMemberModel{}
return groupMember, utils.Wrap(g.DB.Where("group_id = ? and user_id = ?", groupID, userID).Take(groupMember).Error, "") return groupMember, utils.Wrap(g.db(ctx).Where("group_id = ? and user_id = ?", groupID, userID).Take(groupMember).Error, "")
} }
func (g *GroupMemberGorm) TakeOwner(ctx context.Context, groupID string) (groupMember *relation.GroupMemberModel, err error) { func (g *GroupMemberGorm) TakeOwner(ctx context.Context, groupID string) (groupMember *relation.GroupMemberModel, err error) {
groupMember = &relation.GroupMemberModel{} groupMember = &relation.GroupMemberModel{}
return groupMember, utils.Wrap(g.DB.Where("group_id = ? and role_level = ?", groupID, constant.GroupOwner).Take(groupMember).Error, "") return groupMember, utils.Wrap(g.db(ctx).Where("group_id = ? and role_level = ?", groupID, constant.GroupOwner).Take(groupMember).Error, "")
} }
func (g *GroupMemberGorm) SearchMember(ctx context.Context, keyword string, groupIDs []string, userIDs []string, roleLevels []int32, pageNumber, showNumber int32) (total uint32, groupList []*relation.GroupMemberModel, err error) { func (g *GroupMemberGorm) SearchMember(ctx context.Context, keyword string, groupIDs []string, userIDs []string, roleLevels []int32, pageNumber, showNumber int32) (total uint32, groupList []*relation.GroupMemberModel, err error) {
db := g.DB db := g.db(ctx)
ormutil.GormIn(&db, "group_id", groupIDs) ormutil.GormIn(&db, "group_id", groupIDs)
ormutil.GormIn(&db, "user_id", userIDs) ormutil.GormIn(&db, "user_id", userIDs)
ormutil.GormIn(&db, "role_level", roleLevels) ormutil.GormIn(&db, "role_level", roleLevels)
@ -80,7 +80,7 @@ func (g *GroupMemberGorm) SearchMember(ctx context.Context, keyword string, grou
} }
func (g *GroupMemberGorm) MapGroupMemberNum(ctx context.Context, groupIDs []string) (count map[string]uint32, err error) { func (g *GroupMemberGorm) MapGroupMemberNum(ctx context.Context, groupIDs []string) (count map[string]uint32, err error) {
return ormutil.MapCount(g.DB.Where("group_id in (?)", groupIDs), "group_id") return ormutil.MapCount(g.db(ctx).Where("group_id in (?)", groupIDs), "group_id")
} }
func (g *GroupMemberGorm) FindJoinUserID(ctx context.Context, groupIDs []string) (groupUsers map[string][]string, err error) { func (g *GroupMemberGorm) FindJoinUserID(ctx context.Context, groupIDs []string) (groupUsers map[string][]string, err error) {
@ -88,7 +88,7 @@ func (g *GroupMemberGorm) FindJoinUserID(ctx context.Context, groupIDs []string)
GroupID string `gorm:"group_id"` GroupID string `gorm:"group_id"`
UserID string `gorm:"user_id"` UserID string `gorm:"user_id"`
} }
if err := g.DB.Model(&relation.GroupMemberModel{}).Where("group_id in (?)", groupIDs).Find(&items).Error; err != nil { if err := g.db(ctx).Model(&relation.GroupMemberModel{}).Where("group_id in (?)", groupIDs).Find(&items).Error; err != nil {
return nil, utils.Wrap(err, "") return nil, utils.Wrap(err, "")
} }
groupUsers = make(map[string][]string) groupUsers = make(map[string][]string)
@ -99,13 +99,13 @@ func (g *GroupMemberGorm) FindJoinUserID(ctx context.Context, groupIDs []string)
} }
func (g *GroupMemberGorm) FindMemberUserID(ctx context.Context, groupID string) (userIDs []string, err error) { func (g *GroupMemberGorm) FindMemberUserID(ctx context.Context, groupID string) (userIDs []string, err error) {
return userIDs, utils.Wrap(g.DB.Model(&relation.GroupMemberModel{}).Where("group_id = ?", groupID).Pluck("user_id", &userIDs).Error, "") return userIDs, utils.Wrap(g.db(ctx).Model(&relation.GroupMemberModel{}).Where("group_id = ?", groupID).Pluck("user_id", &userIDs).Error, "")
} }
func (g *GroupMemberGorm) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) { func (g *GroupMemberGorm) FindUserJoinedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) {
return groupIDs, utils.Wrap(g.DB.Model(&relation.GroupMemberModel{}).Where("user_id = ?", userID).Pluck("group_id", &groupIDs).Error, "") return groupIDs, utils.Wrap(g.db(ctx).Model(&relation.GroupMemberModel{}).Where("user_id = ?", userID).Pluck("group_id", &groupIDs).Error, "")
} }
func (g *GroupMemberGorm) TakeGroupMemberNum(ctx context.Context, groupID string) (count int64, err error) { func (g *GroupMemberGorm) TakeGroupMemberNum(ctx context.Context, groupID string) (count int64, err error) {
return count, utils.Wrap(g.DB.Model(&relation.GroupMemberModel{}).Where("group_id = ?", groupID).Count(&count).Error, "") return count, utils.Wrap(g.db(ctx).Model(&relation.GroupMemberModel{}).Where("group_id = ?", groupID).Count(&count).Error, "")
} }