IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-10 20:15:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

msgDestruct

Browse Source
This commit is contained in:
wangchuxiao
2023-07-10 10:09:51 +08:00
parent c64657d0b7
commit 7bf9d5a305
19 changed files with 508 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/tools/conversation.go
+33
View File
@@ -0,0 +1,33 @@
package tools
import (
"time"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
)
func (c *MsgTool) ConversationsDestructMsgs() {
ctx := mcontext.NewCtx(utils.GetSelfFuncName())
conversations, err := c.conversationDatabase.GetConversationIDsNeedDestruct(ctx)
if err != nil {
log.ZError(ctx, "get conversation id need destruct failed", err)
return
}
for _, conversation := range conversations {
log.ZDebug(ctx, "UserMsgsDestruct", "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID, "msgDestructTime", conversation.MsgDestructTime, "lastMsgDestructTime", conversation.LatestMsgDestructTime)
seqs, err := c.msgDatabase.UserMsgsDestruct(ctx, conversation.OwnerUserID, conversation.ConversationID, conversation.MsgDestructTime, conversation.LatestMsgDestructTime)
if err != nil {
log.ZError(ctx, "user msg destruct failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
continue
}
if err := c.conversationDatabase.UpdateUsersConversationFiled(ctx, []string{conversation.OwnerUserID}, conversation.ConversationID, map[string]interface{}{"latest_msg_destruct_time": time.Now()}); err != nil {
log.ZError(ctx, "updateUsersConversationFiled failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
continue
}
if err := c.msgNotificationSender.UserDeleteMsgsNotification(ctx, conversation.OwnerUserID, conversation.ConversationID, seqs); err != nil {
log.ZError(ctx, "userDeleteMsgsNotification failed", err, "conversationID", conversation.ConversationID, "ownerUserID", conversation.OwnerUserID)
}
}
}
internal/tools/cron_task.go
+7 -2
View File
@@ -22,8 +22,13 @@ func StartCronTask() error {
wg.Add(1)
_, err = c.AddFunc(config.Config.ChatRecordsClearTime, msgTool.AllConversationClearMsgAndFixSeq)
if err != nil {
fmt.Println("start cron failed", err.Error(), config.Config.ChatRecordsClearTime)
return err
fmt.Println("start allConversationClearMsgAndFixSeq cron failed", err.Error(), config.Config.ChatRecordsClearTime)
panic(err)
}
_, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs)
if err != nil {
fmt.Println("start conversationsDestructMsgs cron failed", err.Error(), config.Config.ChatRecordsClearTime)
panic(err)
}
c.Start()
wg.Wait()
internal/tools/msg.go
+25 -11
View File
@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"math"
"time"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
@@ -14,24 +15,30 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
"github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry/zookeeper"
"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient/notification"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
)
type MsgTool struct {
msgDatabase controller.CommonMsgDatabase
conversationDatabase controller.ConversationDatabase
userDatabase controller.UserDatabase
groupDatabase controller.GroupDatabase
msgDatabase controller.CommonMsgDatabase
conversationDatabase controller.ConversationDatabase
userDatabase controller.UserDatabase
groupDatabase controller.GroupDatabase
msgNotificationSender *notification.MsgNotificationSender
}
var errSeq = errors.New("cache max seq and mongo max seq is diff > 10")
func NewMsgTool(msgDatabase controller.CommonMsgDatabase, userDatabase controller.UserDatabase, groupDatabase controller.GroupDatabase, conversationDatabase controller.ConversationDatabase) *MsgTool {
func NewMsgTool(msgDatabase controller.CommonMsgDatabase, userDatabase controller.UserDatabase,
groupDatabase controller.GroupDatabase, conversationDatabase controller.ConversationDatabase, msgNotificationSender *notification.MsgNotificationSender) *MsgTool {
return &MsgTool{
msgDatabase: msgDatabase,
userDatabase: userDatabase,
groupDatabase: groupDatabase,
conversationDatabase: conversationDatabase,
msgDatabase: msgDatabase,
userDatabase: userDatabase,
groupDatabase: groupDatabase,
conversationDatabase: conversationDatabase,
msgNotificationSender: msgNotificationSender,
}
}
@@ -48,12 +55,20 @@ func InitMsgTool() (*MsgTool, error) {
if err != nil {
return nil, err
}
discov, err := zookeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
zookeeper.WithFreq(time.Hour), zookeeper.WithRoundRobin(), zookeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,
config.Config.Zookeeper.Password), zookeeper.WithTimeout(10), zookeeper.WithLogger(log.NewZkLogger()))
if err != nil {
return nil, err
}
userDB := relation.NewUserGorm(db)
msgDatabase := controller.InitCommonMsgDatabase(rdb, mongo.GetDatabase())
userDatabase := controller.NewUserDatabase(userDB, cache.NewUserCacheRedis(rdb, relation.NewUserGorm(db), cache.GetDefaultOpt()), tx.NewGorm(db))
groupDatabase := controller.InitGroupDatabase(db, rdb, mongo.GetDatabase())
conversationDatabase := controller.NewConversationDatabase(relation.NewConversationGorm(db), cache.NewConversationRedis(rdb, cache.GetDefaultOpt(), relation.NewConversationGorm(db)), tx.NewGorm(db))
msgTool := NewMsgTool(msgDatabase, userDatabase, groupDatabase, conversationDatabase)
msgRpcClient := rpcclient.NewMessageRpcClient(discov)
msgNotificationSender := notification.NewMsgNotificationSender(rpcclient.WithRpcClient(&msgRpcClient))
msgTool := NewMsgTool(msgDatabase, userDatabase, groupDatabase, conversationDatabase, msgNotificationSender)
return msgTool, nil
}
@@ -80,7 +95,6 @@ func (c *MsgTool) ClearConversationsMsg(ctx context.Context, conversationIDs []s
if err := c.checkMaxSeq(ctx, conversationID); err != nil {
log.ZError(ctx, "fixSeq failed", err, "conversationID", conversationID)
}
}
}