@ -24,6 +24,7 @@ import (
kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
pbconversation "github.com/openimsdk/protocol/conversation"
pbconversation "github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/third"
"github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/mw"
"github.com/openimsdk/tools/mw"
@ -58,10 +59,10 @@ func Start(ctx context.Context, config *CronTaskConfig) error {
return err
return err
}
}
// thirdConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Third)
thirdConn , err := client . GetConn ( ctx , config . Share . RpcRegisterName . Third )
// if err != nil {
if err != nil {
//
// }
}
conversationConn , err := client . GetConn ( ctx , config . Share . RpcRegisterName . Conversation )
conversationConn , err := client . GetConn ( ctx , config . Share . RpcRegisterName . Conversation )
if err != nil {
if err != nil {
@ -70,65 +71,82 @@ func Start(ctx context.Context, config *CronTaskConfig) error {
msgClient := msg . NewMsgClient ( msgConn )
msgClient := msg . NewMsgClient ( msgConn )
conversationClient := pbconversation . NewConversationClient ( conversationConn )
conversationClient := pbconversation . NewConversationClient ( conversationConn )
// thirdClient := third.NewThirdClient(thirdConn)
thirdClient := third . NewThirdClient ( thirdConn )
crontab := cron . New ( )
crontab := cron . New ( )
// scheduled hard delete outdated Msgs in specific time.
// scheduled hard delete outdated Msgs in specific time.
clearMsg Func := func ( ) {
destructMsgs Func := func ( ) {
now := time . Now ( )
now := time . Now ( )
deltime := now . Add ( - time . Hour * 24 * time . Duration ( config . CronTask . RetainChatRecords ) )
deltime := now . Add ( - time . Hour * 24 * time . Duration ( config . CronTask . RetainChatRecords ) )
ctx := mcontext . SetOperationID ( ctx , fmt . Sprintf ( "cron_%d_%d" , os . Getpid ( ) , deltime . UnixMilli ( ) ) )
ctx := mcontext . SetOperationID ( ctx , fmt . Sprintf ( "cron_%d_%d" , os . Getpid ( ) , deltime . UnixMilli ( ) ) )
log . ZDebug ( ctx , " clear chat records", "deltime" , deltime , "timestamp" , deltime . UnixMilli ( ) )
log . ZDebug ( ctx , " Destruct chat records", "deltime" , deltime , "timestamp" , deltime . UnixMilli ( ) )
if _ , err := msgClient . ClearMsg( ctx , & msg . ClearMsg Req{ Timestamp : deltime . UnixMilli ( ) } ) ; err != nil {
if _ , err := msgClient . DestructMsgs( ctx , & msg . DestructMsgs Req{ Timestamp : deltime . UnixMilli ( ) } ) ; err != nil {
log . ZError ( ctx , "cron clear chat records failed", err , "deltime" , deltime , "cont" , time . Since ( now ) )
log . ZError ( ctx , "cron destruct chat records failed", err , "deltime" , deltime , "cont" , time . Since ( now ) )
return
return
}
}
log . ZDebug ( ctx , "cron clear chat records success", "deltime" , deltime , "cont" , time . Since ( now ) )
log . ZDebug ( ctx , "cron destruct chat records success", "deltime" , deltime , "cont" , time . Since ( now ) )
}
}
if _ , err := crontab . AddFunc ( config . CronTask . CronExecuteTime , clearMsg Func) ; err != nil {
if _ , err := crontab . AddFunc ( config . CronTask . CronExecuteTime , destructMsgs Func) ; err != nil {
return errs . Wrap ( err )
return errs . Wrap ( err )
}
}
// scheduled soft delete outdated Msgs in specific time when user set `is_msg_destruct` feature.
// scheduled soft delete outdated Msgs in specific time when user set `is_msg_destruct` feature.
msgDestruct Func := func ( ) {
clearMsg Func := func ( ) {
now := time . Now ( )
now := time . Now ( )
ctx := mcontext . SetOperationID ( ctx , fmt . Sprintf ( "cron_%d_%d" , os . Getpid ( ) , now . UnixMilli ( ) ) )
ctx := mcontext . SetOperationID ( ctx , fmt . Sprintf ( "cron_%d_%d" , os . Getpid ( ) , now . UnixMilli ( ) ) )
log . ZDebug ( ctx , " msg destruct cron start", "now" , now )
log . ZDebug ( ctx , " clear msg cron start", "now" , now )
conversations , err := conversationClient . GetConversationsNeed DestructMsgs ( ctx , & pbconversation . GetConversationsNeed DestructMsgs Req{ } )
conversations , err := conversationClient . GetConversationsNeed ClearMsg ( ctx , & pbconversation . GetConversationsNeed ClearMsg Req{ } )
if err != nil {
if err != nil {
log . ZError ( ctx , "Get conversation need Destruct msgs failed." , err )
log . ZError ( ctx , "Get conversation need Destruct msgs failed." , err )
return
return
} else {
}
_ , err := msgClient . DestructMsgs ( ctx , & msg . DestructMsgsReq { Conversations : conversations . Conversations } )
_ , err = msgClient . ClearMsg ( ctx , & msg . ClearMsgReq { Conversations : conversations . Conversations } )
if err != nil {
if err != nil {
log . ZError ( ctx , " Destruct Msgs failed.", err )
log . ZError ( ctx , " Clear Msg failed.", err )
return
return
}
}
log . ZDebug ( ctx , "clear msg cron task completed" , "cont" , time . Since ( now ) )
}
}
log . ZDebug ( ctx , "msg destruct cron task completed" , "cont" , time . Since ( now ) )
if _ , err := crontab . AddFunc ( config . CronTask . CronExecuteTime , clearMsgFunc ) ; err != nil {
}
if _ , err := crontab . AddFunc ( config . CronTask . CronExecuteTime , msgDestructFunc ) ; err != nil {
return errs . Wrap ( err )
return errs . Wrap ( err )
}
}
// // scheduled delete outdated file Objects and their datas in specific time.
// scheduled delete outdated file Objects and their datas in specific time.
// deleteObjectFunc := func() {
deleteObjectFunc := func ( ) {
// now := time.Now()
now := time . Now ( )
// deleteTime := now.Add(-time.Hour * 24 * time.Duration(config.CronTask.FileExpireTime))
executeNum := 5
// ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deleteTime.UnixMilli()))
// number of pagination. if need modify, need update value in third.DeleteOutdatedData
// log.ZDebug(ctx, "deleteoutDatedData ", "deletetime", deleteTime, "timestamp", deleteTime.UnixMilli())
pageShowNumber := 500
// if _, err := thirdClient.DeleteOutdatedData(ctx, &third.DeleteOutdatedDataReq{ExpireTime: deleteTime.UnixMilli()}); err != nil {
deleteTime := now . Add ( - time . Hour * 24 * time . Duration ( config . CronTask . FileExpireTime ) )
// log.ZError(ctx, "cron deleteoutDatedData failed", err, "deleteTime", deleteTime, "cont", time.Since(now))
ctx := mcontext . SetOperationID ( ctx , fmt . Sprintf ( "cron_%d_%d" , os . Getpid ( ) , deleteTime . UnixMilli ( ) ) )
// return
log . ZDebug ( ctx , "deleteoutDatedData" , "deletetime" , deleteTime , "timestamp" , deleteTime . UnixMilli ( ) )
// }
// log.ZDebug(ctx, "cron deleteoutDatedData success", "deltime", deleteTime, "cont", time.Since(now))
if len ( config . CronTask . DeleteObjectType ) == 0 {
// }
log . ZDebug ( ctx , "cron deleteoutDatedData not type need delete" , "deletetime" , deleteTime , "DeleteObjectType" , config . CronTask . DeleteObjectType , "cont" , time . Since ( now ) )
// if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, deleteObjectFunc); err != nil {
return
// return errs.Wrap(err)
}
// }
for i := 0 ; i < executeNum ; i ++ {
resp , err := thirdClient . DeleteOutdatedData ( ctx , & third . DeleteOutdatedDataReq { ExpireTime : deleteTime . UnixMilli ( ) , ObjectGroup : config . CronTask . DeleteObjectType } )
if err != nil {
log . ZError ( ctx , "cron deleteoutDatedData failed" , err , "deleteTime" , deleteTime , "cont" , time . Since ( now ) )
return
}
if resp . Count == 0 || resp . Count < int32 ( pageShowNumber ) {
break
}
}
log . ZDebug ( ctx , "cron deleteoutDatedData success" , "deltime" , deleteTime , "cont" , time . Since ( now ) )
}
if _ , err := crontab . AddFunc ( config . CronTask . CronExecuteTime , deleteObjectFunc ) ; err != nil {
return errs . Wrap ( err )
}
log . ZDebug ( ctx , "start cron task" , "CronExecuteTime" , config . CronTask . CronExecuteTime )
log . ZDebug ( ctx , "start cron task" , "CronExecuteTime" , config . CronTask . CronExecuteTime )
crontab . Start ( )
crontab . Start ( )
icey-yu
10 months ago