merge v3dev into main (#504)

* statistics user register

* refactor: router change

* minio init

* UserRegisterCount

* push use local conn

* refactor: user pb update

* remove online push close grpc conn

* refactor: user pb update

* refactor:pb file

* msgs statistics

* msgs statistics

* revoke userID

* refactor: errcode update

* active user

* active user

* active user

* refactor: errcode update

* feat: conn update token

* active user

* active user

* feat: conn update token

* active user

* feat: conn update token

* feat: conn update token

* feat: conn update token

* add tx_oss cos

* active user

* active user

* group create

* group create

* feat: group notification show to conversation

* feat: group notification show to conversation

* group active

* user active

* sendNotificationWithName

* withname

* privateChat

* a2r call option

* grpc with detail return error

* change log error

* chain unary interceptor

* api nil slice map

* fix sync has read

* fix: text update

* fix: update add model

* set conversations update

* set privateChat

* fix: content update

* remove unuse rpc

* msgDestruct

* cron use rpc mw

* set IsMsgDestruct

* msg destruct

* msgDestruct

* s3 minio, cos, oss support

* feat: add implement of GetUsersOnlineStatus, #472 (#477)

* s3 minio, cos, oss support

* s3 route

* remove extendMsg code

* s3 route

* remove unuse code

* s3 pb

* s3 pb

* s3 pb

* s3 presigned put

* s3 presigned test

* s3 presigned test

* s3 presigned test

* s3 presigned test

* s3 presigned test

* s3 presigned test

* s3 presigned test

* s3 presigned test

* Update .gitignore (#482)

* s3 debug log

* s3 debug log

* cron add log and fix cron

* add log

* cron

* s3 config

* fix kick user bug

* s3 cos

* add kick log

* s3 cos test

* s3 cos test

* s3 cos test

* kick user log

* kickuserlog

* s3 cos copy

* s3 cos copy

* s3 url

* s3 url

* s3 AccessURL

* log

* s3 InitiateMultipartUpload add ExpireTime

---------

Co-authored-by: withchao <993506633@qq.com>
Co-authored-by: wangchuxiao <wangchuxiao97@outlook.com>
Co-authored-by: BanTanger <88583317+BanTanger@users.noreply.github.com>
Co-authored-by: withchao <48119764+withchao@users.noreply.github.com>
Co-authored-by: Alan <68671759+hanzhixiao@users.noreply.github.com>

pkg/common/db/controller/conversation.go

@@ -1,21 +1,8 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package controller
 
 import (
 	"context"
+	"time"
 
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
@@ -27,22 +14,13 @@ import (
 
 type ConversationDatabase interface {
 	//UpdateUserConversationFiled 更新用户该会话的属性信息
-	UpdateUsersConversationFiled(
-		ctx context.Context,
-		userIDs []string,
-		conversationID string,
-		args map[string]interface{},
-	) error
+	UpdateUsersConversationFiled(ctx context.Context, userIDs []string, conversationID string, args map[string]interface{}) error
 	//CreateConversation 创建一批新的会话
 	CreateConversation(ctx context.Context, conversations []*relationTb.ConversationModel) error
 	//SyncPeerUserPrivateConversation 同步对端私聊会话内部保证事务操作
 	SyncPeerUserPrivateConversationTx(ctx context.Context, conversation []*relationTb.ConversationModel) error
 	//FindConversations 根据会话ID获取某个用户的多个会话
-	FindConversations(
-		ctx context.Context,
-		ownerUserID string,
-		conversationIDs []string,
-	) ([]*relationTb.ConversationModel, error)
+	FindConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*relationTb.ConversationModel, error)
 	//FindRecvMsgNotNotifyUserIDs 获取超级大群开启免打扰的用户ID
 	FindRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error)
 	//GetUserAllConversation 获取一个用户在服务器上所有的会话
@@ -50,28 +28,17 @@ type ConversationDatabase interface {
 	//SetUserConversations 设置用户多个会话属性，如果会话不存在则创建，否则更新,内部保证原子性
 	SetUserConversations(ctx context.Context, ownerUserID string, conversations []*relationTb.ConversationModel) error
 	//SetUsersConversationFiledTx 设置多个用户会话关于某个字段的更新操作，如果会话不存在则创建，否则更新，内部保证事务操作
-	SetUsersConversationFiledTx(
-		ctx context.Context,
-		userIDs []string,
-		conversation *relationTb.ConversationModel,
-		filedMap map[string]interface{},
-	) error
+	SetUsersConversationFiledTx(ctx context.Context, userIDs []string, conversation *relationTb.ConversationModel, filedMap map[string]interface{}) error
 	CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error
 	GetConversationIDs(ctx context.Context, userID string) ([]string, error)
 	GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error)
 	GetAllConversationIDs(ctx context.Context) ([]string, error)
 	GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
-	GetConversationsByConversationID(
-		ctx context.Context,
-		conversationIDs []string,
-	) ([]*relationTb.ConversationModel, error)
+	GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationTb.ConversationModel, error)
+	GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationTb.ConversationModel, error)
 }
 
-func NewConversationDatabase(
-	conversation relationTb.ConversationModelInterface,
-	cache cache.ConversationCache,
-	tx tx.Tx,
-) ConversationDatabase {
+func NewConversationDatabase(conversation relationTb.ConversationModelInterface, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
 	return &conversationDatabase{
 		conversationDB: conversation,
 		cache:          cache,
@@ -85,12 +52,7 @@ type conversationDatabase struct {
 	tx             tx.Tx
 }
 
-func (c *conversationDatabase) SetUsersConversationFiledTx(
-	ctx context.Context,
-	userIDs []string,
-	conversation *relationTb.ConversationModel,
-	filedMap map[string]interface{},
-) (err error) {
+func (c *conversationDatabase) SetUsersConversationFiledTx(ctx context.Context, userIDs []string, conversation *relationTb.ConversationModel, filedMap map[string]interface{}) (err error) {
 	cache := c.cache.NewCache()
 	if err := c.tx.Transaction(func(tx any) error {
 		conversationTx := c.conversationDB.NewTx(tx)
@@ -113,12 +75,14 @@ func (c *conversationDatabase) SetUsersConversationFiledTx(
 		NotUserIDs := utils.DifferenceString(haveUserIDs, userIDs)
 		log.ZDebug(ctx, "SetUsersConversationFiledTx", "NotUserIDs", NotUserIDs, "haveUserIDs", haveUserIDs, "userIDs", userIDs)
 		var conversations []*relationTb.ConversationModel
+		now := time.Now()
 		for _, v := range NotUserIDs {
 			temp := new(relationTb.ConversationModel)
 			if err := utils.CopyStructFields(temp, conversation); err != nil {
 				return err
 			}
 			temp.OwnerUserID = v
+			temp.CreateTime = now
 			conversations = append(conversations, temp)
 
 		}
@@ -136,12 +100,7 @@ func (c *conversationDatabase) SetUsersConversationFiledTx(
 	return cache.ExecDel(ctx)
 }
 
-func (c *conversationDatabase) UpdateUsersConversationFiled(
-	ctx context.Context,
-	userIDs []string,
-	conversationID string,
-	args map[string]interface{},
-) error {
+func (c *conversationDatabase) UpdateUsersConversationFiled(ctx context.Context, userIDs []string, conversationID string, args map[string]interface{}) error {
 	_, err := c.conversationDB.UpdateByMap(ctx, userIDs, conversationID, args)
 	if err != nil {
 		return err
@@ -149,10 +108,7 @@ func (c *conversationDatabase) UpdateUsersConversationFiled(
 	return c.cache.DelUsersConversation(conversationID, userIDs...).ExecDel(ctx)
 }
 
-func (c *conversationDatabase) CreateConversation(
-	ctx context.Context,
-	conversations []*relationTb.ConversationModel,
-) error {
+func (c *conversationDatabase) CreateConversation(ctx context.Context, conversations []*relationTb.ConversationModel) error {
 	if err := c.conversationDB.Create(ctx, conversations); err != nil {
 		return err
 	}
@@ -165,35 +121,34 @@ func (c *conversationDatabase) CreateConversation(
 	return cache.DelConversationIDs(userIDs...).DelUserConversationIDsHash(userIDs...).ExecDel(ctx)
 }
 
-func (c *conversationDatabase) SyncPeerUserPrivateConversationTx(
-	ctx context.Context,
-	conversations []*relationTb.ConversationModel,
-) error {
+func (c *conversationDatabase) SyncPeerUserPrivateConversationTx(ctx context.Context, conversations []*relationTb.ConversationModel) error {
 	cache := c.cache.NewCache()
 	if err := c.tx.Transaction(func(tx any) error {
 		conversationTx := c.conversationDB.NewTx(tx)
 		for _, conversation := range conversations {
 			for _, v := range [][2]string{{conversation.OwnerUserID, conversation.UserID}, {conversation.UserID, conversation.OwnerUserID}} {
-				haveUserIDs, err := conversationTx.FindUserID(ctx, []string{v[0]}, []string{conversation.ConversationID})
+				ownerUserID := v[0]
+				userID := v[1]
+				haveUserIDs, err := conversationTx.FindUserID(ctx, []string{ownerUserID}, []string{conversation.ConversationID})
 				if err != nil {
 					return err
 				}
 				if len(haveUserIDs) > 0 {
-					_, err := conversationTx.UpdateByMap(ctx, []string{v[0]}, conversation.ConversationID, map[string]interface{}{"is_private_chat": conversation.IsPrivateChat})
+					_, err := conversationTx.UpdateByMap(ctx, []string{ownerUserID}, conversation.ConversationID, map[string]interface{}{"is_private_chat": conversation.IsPrivateChat})
 					if err != nil {
 						return err
 					}
-					cache = cache.DelUsersConversation(conversation.ConversationID, v[0])
+					cache = cache.DelUsersConversation(conversation.ConversationID, ownerUserID)
 				} else {
 					newConversation := *conversation
-					newConversation.OwnerUserID = v[0]
-					newConversation.UserID = v[1]
+					newConversation.OwnerUserID = ownerUserID
+					newConversation.UserID = userID
 					newConversation.ConversationID = conversation.ConversationID
 					newConversation.IsPrivateChat = conversation.IsPrivateChat
 					if err := conversationTx.Create(ctx, []*relationTb.ConversationModel{&newConversation}); err != nil {
 						return err
 					}
-					cache = cache.DelConversationIDs(v[0]).DelUserConversationIDsHash(v[0])
+					cache = cache.DelConversationIDs(ownerUserID).DelUserConversationIDsHash(ownerUserID)
 				}
 			}
 		}
@@ -201,37 +156,22 @@ func (c *conversationDatabase) SyncPeerUserPrivateConversationTx(
 	}); err != nil {
 		return err
 	}
-	return c.cache.ExecDel(ctx)
+	return cache.ExecDel(ctx)
 }
 
-func (c *conversationDatabase) FindConversations(
-	ctx context.Context,
-	ownerUserID string,
-	conversationIDs []string,
-) ([]*relationTb.ConversationModel, error) {
+func (c *conversationDatabase) FindConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*relationTb.ConversationModel, error) {
 	return c.cache.GetConversations(ctx, ownerUserID, conversationIDs)
 }
 
-func (c *conversationDatabase) GetConversation(
-	ctx context.Context,
-	ownerUserID string,
-	conversationID string,
-) (*relationTb.ConversationModel, error) {
+func (c *conversationDatabase) GetConversation(ctx context.Context, ownerUserID string, conversationID string) (*relationTb.ConversationModel, error) {
 	return c.cache.GetConversation(ctx, ownerUserID, conversationID)
 }
 
-func (c *conversationDatabase) GetUserAllConversation(
-	ctx context.Context,
-	ownerUserID string,
-) ([]*relationTb.ConversationModel, error) {
+func (c *conversationDatabase) GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*relationTb.ConversationModel, error) {
 	return c.cache.GetUserAllConversations(ctx, ownerUserID)
 }
 
-func (c *conversationDatabase) SetUserConversations(
-	ctx context.Context,
-	ownerUserID string,
-	conversations []*relationTb.ConversationModel,
-) error {
+func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUserID string, conversations []*relationTb.ConversationModel) error {
 	cache := c.cache.NewCache()
 	if err := c.tx.Transaction(func(tx any) error {
 		var conversationIDs []string
@@ -281,11 +221,7 @@ func (c *conversationDatabase) FindRecvMsgNotNotifyUserIDs(ctx context.Context,
 	return c.cache.GetSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID)
 }
 
-func (c *conversationDatabase) CreateGroupChatConversation(
-	ctx context.Context,
-	groupID string,
-	userIDs []string,
-) error {
+func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error {
 	cache := c.cache.NewCache()
 	conversationID := utils.GetConversationIDBySessionType(constant.SuperGroupChatType, groupID)
 	if err := c.tx.Transaction(func(tx any) error {
@@ -325,10 +261,7 @@ func (c *conversationDatabase) GetConversationIDs(ctx context.Context, userID st
 	return c.cache.GetUserConversationIDs(ctx, userID)
 }
 
-func (c *conversationDatabase) GetUserConversationIDsHash(
-	ctx context.Context,
-	ownerUserID string,
-) (hash uint64, err error) {
+func (c *conversationDatabase) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) {
 	return c.cache.GetUserConversationIDsHash(ctx, ownerUserID)
 }
 
@@ -336,16 +269,14 @@ func (c *conversationDatabase) GetAllConversationIDs(ctx context.Context) ([]str
 	return c.conversationDB.GetAllConversationIDs(ctx)
 }
 
-func (c *conversationDatabase) GetUserAllHasReadSeqs(
-	ctx context.Context,
-	ownerUserID string,
-) (map[string]int64, error) {
+func (c *conversationDatabase) GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) {
 	return c.cache.GetUserAllHasReadSeqs(ctx, ownerUserID)
 }
 
-func (c *conversationDatabase) GetConversationsByConversationID(
-	ctx context.Context,
-	conversationIDs []string,
-) ([]*relationTb.ConversationModel, error) {
+func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationTb.ConversationModel, error) {
 	return c.conversationDB.GetConversationsByConversationID(ctx, conversationIDs)
 }
+
+func (c *conversationDatabase) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationTb.ConversationModel, error) {
+	return c.conversationDB.GetConversationIDsNeedDestruct(ctx)
+}

pkg/common/db/controller/extend_msg.go

@@ -1,158 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package controller
-
-import (
-	"context"
-
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
-	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
-)
-
-// for mongoDB
-type ExtendMsgDatabase interface {
-	CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error
-	GetAllExtendMsgSet(
-		ctx context.Context,
-		ID string,
-		opts *unRelationTb.GetAllExtendMsgSetOpts,
-	) (sets []*unRelationTb.ExtendMsgSetModel, err error)
-	GetExtendMsgSet(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		maxMsgUpdateTime int64,
-	) (*unRelationTb.ExtendMsgSetModel, error)
-	InsertExtendMsg(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		msg *unRelationTb.ExtendMsgModel,
-	) error
-	InsertOrUpdateReactionExtendMsgSet(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		clientMsgID string,
-		msgFirstModifyTime int64,
-		reactionExtensionList map[string]*unRelationTb.KeyValueModel,
-	) error
-	DeleteReactionExtendMsgSet(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		clientMsgID string,
-		msgFirstModifyTime int64,
-		reactionExtensionList map[string]*unRelationTb.KeyValueModel,
-	) error
-	GetExtendMsg(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		clientMsgID string,
-		maxMsgUpdateTime int64,
-	) (extendMsg *unRelationTb.ExtendMsgModel, err error)
-}
-
-type extendMsgDatabase struct {
-	database unRelationTb.ExtendMsgSetModelInterface
-	cache    cache.ExtendMsgSetCache
-	ctxTx    tx.CtxTx
-}
-
-func NewExtendMsgDatabase(
-	extendMsgModel unRelationTb.ExtendMsgSetModelInterface,
-	cache cache.ExtendMsgSetCache,
-	ctxTx tx.CtxTx,
-) ExtendMsgDatabase {
-	return &extendMsgDatabase{database: extendMsgModel, cache: cache, ctxTx: ctxTx}
-}
-
-func (e *extendMsgDatabase) CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error {
-	return e.database.CreateExtendMsgSet(ctx, set)
-}
-
-func (e *extendMsgDatabase) GetAllExtendMsgSet(
-	ctx context.Context,
-	conversationID string,
-	opts *unRelationTb.GetAllExtendMsgSetOpts,
-) (sets []*unRelationTb.ExtendMsgSetModel, err error) {
-	return e.database.GetAllExtendMsgSet(ctx, conversationID, opts)
-}
-
-func (e *extendMsgDatabase) GetExtendMsgSet(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	maxMsgUpdateTime int64,
-) (*unRelationTb.ExtendMsgSetModel, error) {
-	return e.database.GetExtendMsgSet(ctx, conversationID, sessionType, maxMsgUpdateTime)
-}
-
-func (e *extendMsgDatabase) InsertExtendMsg(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	msg *unRelationTb.ExtendMsgModel,
-) error {
-	return e.database.InsertExtendMsg(ctx, conversationID, sessionType, msg)
-}
-
-func (e *extendMsgDatabase) InsertOrUpdateReactionExtendMsgSet(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	clientMsgID string,
-	msgFirstModifyTime int64,
-	reactionExtensionList map[string]*unRelationTb.KeyValueModel,
-) error {
-	return e.database.InsertOrUpdateReactionExtendMsgSet(
-		ctx,
-		conversationID,
-		sessionType,
-		clientMsgID,
-		msgFirstModifyTime,
-		reactionExtensionList,
-	)
-}
-
-func (e *extendMsgDatabase) DeleteReactionExtendMsgSet(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	clientMsgID string,
-	msgFirstModifyTime int64,
-	reactionExtensionList map[string]*unRelationTb.KeyValueModel,
-) error {
-	return e.database.DeleteReactionExtendMsgSet(
-		ctx,
-		conversationID,
-		sessionType,
-		clientMsgID,
-		msgFirstModifyTime,
-		reactionExtensionList,
-	)
-}
-
-func (e *extendMsgDatabase) GetExtendMsg(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	clientMsgID string,
-	maxMsgUpdateTime int64,
-) (extendMsg *unRelationTb.ExtendMsgModel, err error) {
-	return e.cache.GetExtendMsg(ctx, conversationID, sessionType, clientMsgID, maxMsgUpdateTime)
-}

pkg/common/db/controller/group.go

@@ -17,6 +17,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/dtm-labs/rockscache"
 	"github.com/redis/go-redis/v9"
@@ -121,6 +122,11 @@ type GroupDatabase interface {
 	DeleteSuperGroup(ctx context.Context, groupID string) error
 	DeleteSuperGroupMember(ctx context.Context, groupID string, userIDs []string) error
 	CreateSuperGroupMember(ctx context.Context, groupID string, userIDs []string) error
+
+	// 获取群总数
+	CountTotal(ctx context.Context, before *time.Time) (count int64, err error)
+	// 获取范围内群增量
+	CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error)
 }
 
 func NewGroupDatabase(
@@ -562,3 +568,11 @@ func (g *groupDatabase) CreateSuperGroupMember(ctx context.Context, groupID stri
 	}
 	return g.cache.DelSuperGroupMemberIDs(groupID).DelJoinedSuperGroupIDs(userIDs...).ExecDel(ctx)
 }
+
+func (g *groupDatabase) CountTotal(ctx context.Context, before *time.Time) (count int64, err error) {
+	return g.groupDB.CountTotal(ctx, before)
+}
+
+func (g *groupDatabase) CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error) {
+	return g.groupDB.CountRangeEverydayTotal(ctx, start, end)
+}

pkg/common/db/controller/msg.go

@@ -1,21 +1,6 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package controller
 
 import (
-	"fmt"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -33,11 +18,10 @@ import (
 	"context"
 	"errors"
 
-	"go.mongodb.org/mongo-driver/mongo"
-
 	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"go.mongodb.org/mongo-driver/mongo"
 )
 
 const (
@@ -56,28 +40,17 @@ type CommonMsgDatabase interface {
 	DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
 	DelUserDeleteMsgsList(ctx context.Context, conversationID string, seqs []int64)
 	// incrSeq然后批量插入缓存
-	BatchInsertChat2Cache(
-		ctx context.Context,
-		conversationID string,
-		msgs []*sdkws.MsgData,
-	) (seq int64, isNewConversation bool, err error)
+	BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error)
 
 	//  通过seqList获取mongo中写扩散消息
-	GetMsgBySeqsRange(
-		ctx context.Context,
-		userID string,
-		conversationID string,
-		begin, end, num, userMaxSeq int64,
-	) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
+	GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
 	// 通过seqList获取大群在 mongo里面的消息
-	GetMsgBySeqs(
-		ctx context.Context,
-		userID string,
-		conversationID string,
-		seqs []int64,
-	) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
+	GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
 	// 删除会话消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache)
 	DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
+	// 用户标记删除过期消息返回标记删除的seq列表
+	UserMsgsDestruct(cte context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error)
+
 	// 用户根据seq删除消息
 	DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error
 	// 物理删除消息置空
@@ -101,11 +74,8 @@ type CommonMsgDatabase interface {
 	GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error)
 	UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error
 
-	GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (maxSeq, minSeq int64, err error)
-	GetConversationMinMaxSeqInMongoAndCache(
-		ctx context.Context,
-		conversationID string,
-	) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
+	GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error)
+	GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
 	SetSendMsgStatus(ctx context.Context, id string, status int32) error
 	GetSendMsgStatus(ctx context.Context, id string) (int32, error)
 
@@ -115,51 +85,17 @@ type CommonMsgDatabase interface {
 	MsgToPushMQ(ctx context.Context, key, conversarionID string, msg2mq *sdkws.MsgData) (int32, int64, error)
 	MsgToMongoMQ(ctx context.Context, key, conversarionID string, msgs []*sdkws.MsgData, lastSeq int64) error
 
-	// modify
-	JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error)
-	SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error
-	SetMessageReactionExpire(
-		ctx context.Context,
-		clientMsgID string,
-		sessionType int32,
-		expiration time.Duration,
-	) (bool, error)
-	GetExtendMsg(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		clientMsgID string,
-		maxMsgUpdateTime int64,
-	) (*pbMsg.ExtendMsg, error)
-	InsertOrUpdateReactionExtendMsgSet(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		clientMsgID string,
-		msgFirstModifyTime int64,
-		reactionExtensionList map[string]*sdkws.KeyValue,
-	) error
-	GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error)
-	GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error)
-	DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error
-	DeleteReactionExtendMsgSet(
-		ctx context.Context,
-		conversationID string,
-		sessionType int32,
-		clientMsgID string,
-		msgFirstModifyTime int64,
-		reactionExtensionList map[string]*sdkws.KeyValue,
-	) error
+	RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error)
+	RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error)
 }
 
 func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase {
 	return &commonMsgDatabase{
-		msgDocDatabase:   msgDocModel,
-		cache:            cacheModel,
-		producer:         kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.LatestMsgToRedis.Topic),
-		producerToMongo:  kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToMongo.Topic),
-		producerToPush:   kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToPush.Topic),
-		producerToModify: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToModify.Topic),
+		msgDocDatabase:  msgDocModel,
+		cache:           cacheModel,
+		producer:        kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.LatestMsgToRedis.Topic),
+		producerToMongo: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToMongo.Topic),
+		producerToPush:  kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToPush.Topic),
 	}
 }
 
@@ -171,15 +107,13 @@ func InitCommonMsgDatabase(rdb redis.UniversalClient, database *mongo.Database)
 }
 
 type commonMsgDatabase struct {
-	msgDocDatabase    unRelationTb.MsgDocModelInterface
-	extendMsgDatabase unRelationTb.ExtendMsgSetModelInterface
-	extendMsgSetModel unRelationTb.ExtendMsgSetModel
-	msg               unRelationTb.MsgDocModel
-	cache             cache.MsgModel
-	producer          *kafka.Producer
-	producerToMongo   *kafka.Producer
-	producerToModify  *kafka.Producer
-	producerToPush    *kafka.Producer
+	msgDocDatabase   unRelationTb.MsgDocModelInterface
+	msg              unRelationTb.MsgDocModel
+	cache            cache.MsgModel
+	producer         *kafka.Producer
+	producerToMongo  *kafka.Producer
+	producerToModify *kafka.Producer
+	producerToPush   *kafka.Producer
 }
 
 func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error {
@@ -187,32 +121,16 @@ func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sd
 	return err
 }
 
-func (db *commonMsgDatabase) MsgToModifyMQ(
-	ctx context.Context,
-	key, conversationID string,
-	messages []*sdkws.MsgData,
-) error {
+func (db *commonMsgDatabase) MsgToModifyMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData) error {
 	if len(messages) > 0 {
-		_, _, err := db.producerToModify.SendMessage(
-			ctx,
-			key,
-			&pbMsg.MsgDataToModifyByMQ{ConversationID: conversationID, Messages: messages},
-		)
+		_, _, err := db.producerToModify.SendMessage(ctx, key, &pbMsg.MsgDataToModifyByMQ{ConversationID: conversationID, Messages: messages})
 		return err
 	}
 	return nil
 }
 
-func (db *commonMsgDatabase) MsgToPushMQ(
-	ctx context.Context,
-	key, conversationID string,
-	msg2mq *sdkws.MsgData,
-) (int32, int64, error) {
-	partition, offset, err := db.producerToPush.SendMessage(
-		ctx,
-		key,
-		&pbMsg.PushMsgDataToMQ{MsgData: msg2mq, ConversationID: conversationID},
-	)
+func (db *commonMsgDatabase) MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error) {
+	partition, offset, err := db.producerToPush.SendMessage(ctx, key, &pbMsg.PushMsgDataToMQ{MsgData: msg2mq, ConversationID: conversationID})
 	if err != nil {
 		log.ZError(ctx, "MsgToPushMQ", err, "key", key, "msg2mq", msg2mq)
 		return 0, 0, err
@@ -220,30 +138,15 @@ func (db *commonMsgDatabase) MsgToPushMQ(
 	return partition, offset, nil
 }
 
-func (db *commonMsgDatabase) MsgToMongoMQ(
-	ctx context.Context,
-	key, conversationID string,
-	messages []*sdkws.MsgData,
-	lastSeq int64,
-) error {
+func (db *commonMsgDatabase) MsgToMongoMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData, lastSeq int64) error {
 	if len(messages) > 0 {
-		_, _, err := db.producerToMongo.SendMessage(
-			ctx,
-			key,
-			&pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, ConversationID: conversationID, MsgData: messages},
-		)
+		_, _, err := db.producerToMongo.SendMessage(ctx, key, &pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, ConversationID: conversationID, MsgData: messages})
 		return err
 	}
 	return nil
 }
 
-func (db *commonMsgDatabase) BatchInsertBlock(
-	ctx context.Context,
-	conversationID string,
-	fields []any,
-	key int8,
-	firstSeq int64,
-) error {
+func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationID string, fields []any, key int8, firstSeq int64) error {
 	if len(fields) == 0 {
 		return nil
 	}
@@ -344,12 +247,7 @@ func (db *commonMsgDatabase) BatchInsertBlock(
 	return nil
 }
 
-func (db *commonMsgDatabase) BatchInsertChat2DB(
-	ctx context.Context,
-	conversationID string,
-	msgList []*sdkws.MsgData,
-	currentMaxSeq int64,
-) error {
+func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
 	if len(msgList) == 0 {
 		return errs.ErrArgs.Wrap("msgList is empty")
 	}
@@ -395,21 +293,11 @@ func (db *commonMsgDatabase) BatchInsertChat2DB(
 	return db.BatchInsertBlock(ctx, conversationID, msgs, updateKeyMsg, msgList[0].Seq)
 }
 
-func (db *commonMsgDatabase) RevokeMsg(
-	ctx context.Context,
-	conversationID string,
-	seq int64,
-	revoke *unRelationTb.RevokeModel,
-) error {
+func (db *commonMsgDatabase) RevokeMsg(ctx context.Context, conversationID string, seq int64, revoke *unRelationTb.RevokeModel) error {
 	return db.BatchInsertBlock(ctx, conversationID, []any{revoke}, updateKeyRevoke, seq)
 }
 
-func (db *commonMsgDatabase) MarkSingleChatMsgsAsRead(
-	ctx context.Context,
-	userID string,
-	conversationID string,
-	totalSeqs []int64,
-) error {
+func (db *commonMsgDatabase) MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, totalSeqs []int64) error {
 	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, totalSeqs) {
 		var indexes []int64
 		for _, seq := range seqs {
@@ -432,11 +320,7 @@ func (db *commonMsgDatabase) DelUserDeleteMsgsList(ctx context.Context, conversa
 	db.cache.DelUserDeleteMsgsList(ctx, conversationID, seqs)
 }
 
-func (db *commonMsgDatabase) BatchInsertChat2Cache(
-	ctx context.Context,
-	conversationID string,
-	msgs []*sdkws.MsgData,
-) (seq int64, isNew bool, err error) {
+func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
 	currentMaxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		prome.Inc(prome.SeqGetFailedCounter)
@@ -483,11 +367,7 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(
 	return lastMaxSeq, isNew, utils.Wrap(err, "")
 }
 
-func (db *commonMsgDatabase) getMsgBySeqs(
-	ctx context.Context,
-	userID, conversationID string,
-	seqs []int64,
-) (totalMsgs []*sdkws.MsgData, err error) {
+func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
 	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, seqs) {
 		//log.ZDebug(ctx, "getMsgBySeqs", "docID", docID, "seqs", seqs)
 		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, seqs)
@@ -501,50 +381,7 @@ func (db *commonMsgDatabase) getMsgBySeqs(
 	return totalMsgs, nil
 }
 
-// func (db *commonMsgDatabase) refetchDelSeqsMsgs(ctx context.Context, conversationID string, delNums, rangeBegin,
-// begin int64) (seqMsgs []*unRelationTb.MsgDataModel, err error) {
-// 	var reFetchSeqs []int64
-// 	if delNums > 0 {
-// 		newBeginSeq := rangeBegin - delNums
-// 		if newBeginSeq >= begin {
-// 			newEndSeq := rangeBegin - 1
-// 			for i := newBeginSeq; i <= newEndSeq; i++ {
-// 				reFetchSeqs = append(reFetchSeqs, i)
-// 			}
-// 		}
-// 	}
-// 	if len(reFetchSeqs) == 0 {
-// 		return
-// 	}
-// 	if len(reFetchSeqs) > 0 {
-// 		m := db.msg.GetDocIDSeqsMap(conversationID, reFetchSeqs)
-// 		for docID, seqs := range m {
-// 			msgs, _, err := db.findMsgInfoBySeq(ctx, docID, seqs)
-// 			if err != nil {
-// 				return nil, err
-// 			}
-// 			for _, msg := range msgs {
-// 				if msg.Status != constant.MsgDeleted {
-// 					seqMsgs = append(seqMsgs, msg)
-// 				}
-// 			}
-// 		}
-// 	}
-// 	if len(seqMsgs) < int(delNums) {
-// 		seqMsgs2, err := db.refetchDelSeqsMsgs(ctx, conversationID, delNums-int64(len(seqMsgs)), rangeBegin-1, begin)
-// 		if err != nil {
-// 			return seqMsgs, err
-// 		}
-// 		seqMsgs = append(seqMsgs, seqMsgs2...)
-// 	}
-// 	return seqMsgs, nil
-// }
-
-func (db *commonMsgDatabase) findMsgInfoBySeq(
-	ctx context.Context,
-	userID, docID string,
-	seqs []int64,
-) (totalMsgs []*unRelationTb.MsgInfoModel, err error) {
+func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID string, seqs []int64) (totalMsgs []*unRelationTb.MsgInfoModel, err error) {
 	msgs, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, userID, docID, seqs)
 	for _, msg := range msgs {
 		if msg.IsRead {
@@ -554,25 +391,8 @@ func (db *commonMsgDatabase) findMsgInfoBySeq(
 	return msgs, err
 }
 
-func (db *commonMsgDatabase) getMsgBySeqsRange(
-	ctx context.Context,
-	userID string,
-	conversationID string,
-	allSeqs []int64,
-	begin, end int64,
-) (seqMsgs []*sdkws.MsgData, err error) {
-	log.ZDebug(
-		ctx,
-		"getMsgBySeqsRange",
-		"conversationID",
-		conversationID,
-		"allSeqs",
-		allSeqs,
-		"begin",
-		begin,
-		"end",
-		end,
-	)
+func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID string, conversationID string, allSeqs []int64, begin, end int64) (seqMsgs []*sdkws.MsgData, err error) {
+	log.ZDebug(ctx, "getMsgBySeqsRange", "conversationID", conversationID, "allSeqs", allSeqs, "begin", begin, "end", end)
 	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, allSeqs) {
 		log.ZDebug(ctx, "getMsgBySeqsRange", "docID", docID, "seqs", seqs)
 		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, seqs)
@@ -589,12 +409,7 @@ func (db *commonMsgDatabase) getMsgBySeqsRange(
 	return seqMsgs, nil
 }
 
-func (db *commonMsgDatabase) GetMsgBySeqsRange(
-	ctx context.Context,
-	userID string,
-	conversationID string,
-	begin, end, num, userMaxSeq int64,
-) (int64, int64, []*sdkws.MsgData, error) {
+func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (int64, int64, []*sdkws.MsgData, error) {
 	userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return 0, 0, nil, err
@@ -614,18 +429,7 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return 0, 0, nil, err
 	}
-	log.ZDebug(
-		ctx,
-		"GetMsgBySeqsRange",
-		"userMinSeq",
-		userMinSeq,
-		"conMinSeq",
-		minSeq,
-		"conMaxSeq",
-		maxSeq,
-		"userMaxSeq",
-		userMaxSeq,
-	)
+	log.ZDebug(ctx, "GetMsgBySeqsRange", "userMinSeq", userMinSeq, "conMinSeq", minSeq, "conMaxSeq", maxSeq, "userMaxSeq", userMaxSeq)
 	if userMaxSeq != 0 {
 		if userMaxSeq < maxSeq {
 			maxSeq = userMaxSeq
@@ -675,18 +479,7 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(
 				cacheDelNum += 1
 			}
 		}
-		log.ZDebug(
-			ctx,
-			"get delSeqs from redis",
-			"delSeqs",
-			delSeqs,
-			"userID",
-			userID,
-			"conversationID",
-			conversationID,
-			"cacheDelNum",
-			cacheDelNum,
-		)
+		log.ZDebug(ctx, "get delSeqs from redis", "delSeqs", delSeqs, "userID", userID, "conversationID", conversationID, "cacheDelNum", cacheDelNum)
 		var reGetSeqsCache []int64
 		for i := 1; i <= cacheDelNum; {
 			newSeq := newBegin - int64(i)
@@ -706,15 +499,7 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(
 			if err != nil {
 				if err != redis.Nil {
 					prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs2))
-					log.ZError(
-						ctx,
-						"get message from redis exception",
-						err,
-						"conversationID",
-						conversationID,
-						"seqs",
-						reGetSeqsCache,
-					)
+					log.ZError(ctx, "get message from redis exception", err, "conversationID", conversationID, "seqs", reGetSeqsCache)
 				}
 			}
 			failedSeqs = append(failedSeqs, failedSeqs2...)
@@ -740,12 +525,7 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(
 	return minSeq, maxSeq, successMsgs, nil
 }
 
-func (db *commonMsgDatabase) GetMsgBySeqs(
-	ctx context.Context,
-	userID string,
-	conversationID string,
-	seqs []int64,
-) (int64, int64, []*sdkws.MsgData, error) {
+func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
 	userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		return 0, 0, nil, err
@@ -771,33 +551,10 @@ func (db *commonMsgDatabase) GetMsgBySeqs(
 	if err != nil {
 		if err != redis.Nil {
 			prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs))
-			log.ZError(
-				ctx,
-				"get message from redis exception",
-				err,
-				"failedSeqs",
-				failedSeqs,
-				"conversationID",
-				conversationID,
-			)
+			log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
 		}
 	}
-	log.ZInfo(
-		ctx,
-		"db.cache.GetMessagesBySeq",
-		"userID",
-		userID,
-		"conversationID",
-		conversationID,
-		"seqs",
-		seqs,
-		"successMsgs",
-		len(successMsgs),
-		"failedSeqs",
-		failedSeqs,
-		"conversationID",
-		conversationID,
-	)
+	log.ZInfo(ctx, "db.cache.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs", seqs, "successMsgs", len(successMsgs), "failedSeqs", failedSeqs, "conversationID", conversationID)
 	prome.Add(prome.MsgPullFromRedisSuccessCounter, len(successMsgs))
 	if len(failedSeqs) > 0 {
 		mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
@@ -811,11 +568,7 @@ func (db *commonMsgDatabase) GetMsgBySeqs(
 	return minSeq, maxSeq, successMsgs, nil
 }
 
-func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(
-	ctx context.Context,
-	conversationID string,
-	remainTime int64,
-) error {
+func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
 	var delStruct delMsgRecursionStruct
 	var skip int64
 	minSeq, err := db.deleteMsgRecursion(ctx, conversationID, skip, &delStruct, remainTime)
@@ -835,6 +588,49 @@ func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(
 	return db.cache.SetMinSeq(ctx, conversationID, minSeq)
 }
 
+func (db *commonMsgDatabase) UserMsgsDestruct(ctx context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error) {
+	var index int64
+	for {
+		// from oldest 2 newest
+		msgDocModel, err := db.msgDocDatabase.GetMsgDocModelByIndex(ctx, conversationID, index, 1)
+		if err != nil || msgDocModel.DocID == "" {
+			if err != nil {
+				if err == unrelation.ErrMsgListNotExist {
+					log.ZDebug(ctx, "deleteMsgRecursion finished", "conversationID", conversationID, "userID", userID, "index", index)
+				} else {
+					log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
+				}
+			}
+			// 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList), 结束递归
+			break
+		}
+		index++
+		//&& msgDocModel.Msg[0].Msg.SendTime > lastMsgDestructTime.UnixMilli()
+		if len(msgDocModel.Msg) > 0 {
+			for _, msg := range msgDocModel.Msg {
+				if msg != nil && msg.Msg != nil && msg.Msg.SendTime+destructTime*1000 <= time.Now().UnixMilli() {
+					if msg.Msg.SendTime > lastMsgDestructTime.UnixMilli() && !utils.Contain(userID, msg.DelList...) {
+						seqs = append(seqs, msg.Msg.Seq)
+					}
+				} else {
+					log.ZDebug(ctx, "deleteMsgRecursion finished", "conversationID", conversationID, "userID", userID, "index", index)
+					break
+				}
+
+			}
+		}
+	}
+
+	log.ZDebug(ctx, "UserMsgsDestruct", "conversationID", conversationID, "userID", userID, "seqs", seqs)
+	if len(seqs) > 0 {
+		latestSeq := seqs[len(seqs)-1]
+		if err := db.cache.SetConversationUserMinSeq(ctx, conversationID, userID, latestSeq); err != nil {
+			return nil, err
+		}
+	}
+	return seqs, nil
+}
+
 // this is struct for recursion
 type delMsgRecursionStruct struct {
 	minSeq    int64
@@ -849,26 +645,13 @@ func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
 // seq 70
 // set minSeq 21
 // recursion 删除list并且返回设置的最小seq
-func (db *commonMsgDatabase) deleteMsgRecursion(
-	ctx context.Context,
-	conversationID string,
-	index int64,
-	delStruct *delMsgRecursionStruct,
-	remainTime int64,
-) (int64, error) {
+func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversationID string, index int64, delStruct *delMsgRecursionStruct, remainTime int64) (int64, error) {
 	// find from oldest list
 	msgDocModel, err := db.msgDocDatabase.GetMsgDocModelByIndex(ctx, conversationID, index, 1)
 	if err != nil || msgDocModel.DocID == "" {
 		if err != nil {
 			if err == unrelation.ErrMsgListNotExist {
-				log.ZDebug(
-					ctx,
-					"deleteMsgRecursion ErrMsgListNotExist",
-					"conversationID",
-					conversationID,
-					"index:",
-					index,
-				)
+				log.ZDebug(ctx, "deleteMsgRecursion ErrMsgListNotExist", "conversationID", conversationID, "index:", index)
 			} else {
 				log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
 			}
@@ -880,23 +663,11 @@ func (db *commonMsgDatabase) deleteMsgRecursion(
 		}
 		return delStruct.getSetMinSeq() + 1, nil
 	}
-	log.ZDebug(
-		ctx,
-		"doc info",
-		"conversationID",
-		conversationID,
-		"index",
-		index,
-		"docID",
-		msgDocModel.DocID,
-		"len",
-		len(msgDocModel.Msg),
-	)
+	log.ZDebug(ctx, "doc info", "conversationID", conversationID, "index", index, "docID", msgDocModel.DocID, "len", len(msgDocModel.Msg))
 	if int64(len(msgDocModel.Msg)) > db.msg.GetSingleGocMsgNum() {
 		log.ZWarn(ctx, "msgs too large", nil, "lenth", len(msgDocModel.Msg), "docID:", msgDocModel.DocID)
 	}
-	if msgDocModel.IsFull() &&
-		msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() {
+	if msgDocModel.IsFull() && msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() {
 		log.ZDebug(ctx, "doc is full and all msg is expired", "docID", msgDocModel.DocID)
 		delStruct.delDocIDs = append(delStruct.delDocIDs, msgDocModel.DocID)
 		delStruct.minSeq = msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.Seq
@@ -933,11 +704,7 @@ func (db *commonMsgDatabase) deleteMsgRecursion(
 	return seq, err
 }
 
-func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(
-	ctx context.Context,
-	conversationID string,
-	allSeqs []int64,
-) error {
+func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conversationID string, allSeqs []int64) error {
 	if err := db.cache.DeleteMessages(ctx, conversationID, allSeqs); err != nil {
 		return err
 	}
@@ -953,12 +720,7 @@ func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(
 	return nil
 }
 
-func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(
-	ctx context.Context,
-	userID string,
-	conversationID string,
-	seqs []int64,
-) error {
+func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error {
 	cachedMsgs, _, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs)
 	if err != nil && errs.Unwrap(err) != redis.Nil {
 		log.ZWarn(ctx, "DeleteUserMsgsBySeqs", err, "conversationID", conversationID, "seqs", seqs)
@@ -1027,70 +789,31 @@ func (db *commonMsgDatabase) GetMinSeqs(ctx context.Context, conversationIDs []s
 func (db *commonMsgDatabase) GetMinSeq(ctx context.Context, conversationID string) (int64, error) {
 	return db.cache.GetMinSeq(ctx, conversationID)
 }
-
-func (db *commonMsgDatabase) GetConversationUserMinSeq(
-	ctx context.Context,
-	conversationID string,
-	userID string,
-) (int64, error) {
+func (db *commonMsgDatabase) GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error) {
 	return db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
 }
-
-func (db *commonMsgDatabase) GetConversationUserMinSeqs(
-	ctx context.Context,
-	conversationID string,
-	userIDs []string,
-) (map[string]int64, error) {
+func (db *commonMsgDatabase) GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error) {
 	return db.cache.GetConversationUserMinSeqs(ctx, conversationID, userIDs)
 }
-
-func (db *commonMsgDatabase) SetConversationUserMinSeq(
-	ctx context.Context,
-	conversationID string,
-	userID string,
-	minSeq int64,
-) error {
+func (db *commonMsgDatabase) SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error {
 	return db.cache.SetConversationUserMinSeq(ctx, conversationID, userID, minSeq)
 }
-
-func (db *commonMsgDatabase) SetConversationUserMinSeqs(
-	ctx context.Context,
-	conversationID string,
-	seqs map[string]int64,
-) (err error) {
+func (db *commonMsgDatabase) SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error) {
 	return db.cache.SetConversationUserMinSeqs(ctx, conversationID, seqs)
 }
 
-func (db *commonMsgDatabase) SetUserConversationsMinSeqs(
-	ctx context.Context,
-	userID string,
-	seqs map[string]int64,
-) error {
+func (db *commonMsgDatabase) SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) error {
 	return db.cache.SetUserConversationsMinSeqs(ctx, userID, seqs)
 }
 
-func (db *commonMsgDatabase) UserSetHasReadSeqs(
-	ctx context.Context,
-	userID string,
-	hasReadSeqs map[string]int64,
-) error {
+func (db *commonMsgDatabase) UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error {
 	return db.cache.UserSetHasReadSeqs(ctx, userID, hasReadSeqs)
 }
 
-func (db *commonMsgDatabase) SetHasReadSeq(
-	ctx context.Context,
-	userID string,
-	conversationID string,
-	hasReadSeq int64,
-) error {
+func (db *commonMsgDatabase) SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error {
 	return db.cache.SetHasReadSeq(ctx, userID, conversationID, hasReadSeq)
 }
-
-func (db *commonMsgDatabase) GetHasReadSeqs(
-	ctx context.Context,
-	userID string,
-	conversationIDs []string,
-) (map[string]int64, error) {
+func (db *commonMsgDatabase) GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error) {
 	return db.cache.GetHasReadSeqs(ctx, userID, conversationIDs)
 }
 
@@ -1106,10 +829,7 @@ func (db *commonMsgDatabase) GetSendMsgStatus(ctx context.Context, id string) (i
 	return db.cache.GetSendMsgStatus(ctx, id)
 }
 
-func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(
-	ctx context.Context,
-	conversationID string,
-) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) {
+func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) {
 	minSeqMongo, maxSeqMongo, err = db.GetMinMaxSeqMongo(ctx, conversationID)
 	if err != nil {
 		return
@@ -1125,17 +845,11 @@ func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(
 	return
 }
 
-func (db *commonMsgDatabase) GetMongoMaxAndMinSeq(
-	ctx context.Context,
-	conversationID string,
-) (maxSeq, minSeq int64, err error) {
+func (db *commonMsgDatabase) GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
 	return db.GetMinMaxSeqMongo(ctx, conversationID)
 }
 
-func (db *commonMsgDatabase) GetMinMaxSeqMongo(
-	ctx context.Context,
-	conversationID string,
-) (minSeqMongo, maxSeqMongo int64, err error) {
+func (db *commonMsgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
 	oldestMsgMongo, err := db.msgDocDatabase.GetOldestMsg(ctx, conversationID)
 	if err != nil {
 		return
@@ -1149,124 +863,10 @@ func (db *commonMsgDatabase) GetMinMaxSeqMongo(
 	return
 }
 
-func (db *commonMsgDatabase) JudgeMessageReactionExist(
-	ctx context.Context,
-	clientMsgID string,
-	sessionType int32,
-) (bool, error) {
-	return db.cache.JudgeMessageReactionExist(ctx, clientMsgID, sessionType)
+func (db *commonMsgDatabase) RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error) {
+	return db.msgDocDatabase.RangeUserSendCount(ctx, start, end, group, ase, pageNumber, showNumber)
 }
 
-func (db *commonMsgDatabase) SetMessageTypeKeyValue(
-	ctx context.Context,
-	clientMsgID string,
-	sessionType int32,
-	typeKey, value string,
-) error {
-	return db.cache.SetMessageTypeKeyValue(ctx, clientMsgID, sessionType, typeKey, value)
-}
-
-func (db *commonMsgDatabase) SetMessageReactionExpire(
-	ctx context.Context,
-	clientMsgID string,
-	sessionType int32,
-	expiration time.Duration,
-) (bool, error) {
-	return db.cache.SetMessageReactionExpire(ctx, clientMsgID, sessionType, expiration)
-}
-
-func (db *commonMsgDatabase) GetMessageTypeKeyValue(
-	ctx context.Context,
-	clientMsgID string,
-	sessionType int32,
-	typeKey string,
-) (string, error) {
-	return db.cache.GetMessageTypeKeyValue(ctx, clientMsgID, sessionType, typeKey)
-}
-
-func (db *commonMsgDatabase) GetOneMessageAllReactionList(
-	ctx context.Context,
-	clientMsgID string,
-	sessionType int32,
-) (map[string]string, error) {
-	return db.cache.GetOneMessageAllReactionList(ctx, clientMsgID, sessionType)
-}
-
-func (db *commonMsgDatabase) DeleteOneMessageKey(
-	ctx context.Context,
-	clientMsgID string,
-	sessionType int32,
-	subKey string,
-) error {
-	return db.cache.DeleteOneMessageKey(ctx, clientMsgID, sessionType, subKey)
-}
-
-func (db *commonMsgDatabase) InsertOrUpdateReactionExtendMsgSet(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	clientMsgID string,
-	msgFirstModifyTime int64,
-	reactionExtensions map[string]*sdkws.KeyValue,
-) error {
-	return db.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(
-		ctx,
-		conversationID,
-		sessionType,
-		clientMsgID,
-		msgFirstModifyTime,
-		db.extendMsgSetModel.Pb2Model(reactionExtensions),
-	)
-}
-
-func (db *commonMsgDatabase) GetExtendMsg(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	clientMsgID string,
-	maxMsgUpdateTime int64,
-) (*pbMsg.ExtendMsg, error) {
-	extendMsgSet, err := db.extendMsgDatabase.GetExtendMsgSet(ctx, conversationID, sessionType, maxMsgUpdateTime)
-	if err != nil {
-		return nil, err
-	}
-	extendMsg, ok := extendMsgSet.ExtendMsgs[clientMsgID]
-	if !ok {
-		return nil, errs.ErrRecordNotFound.Wrap(fmt.Sprintf("cant find client msg id: %s", clientMsgID))
-	}
-	reactionExtensionList := make(map[string]*pbMsg.KeyValueResp)
-	for key, model := range extendMsg.ReactionExtensionList {
-		reactionExtensionList[key] = &pbMsg.KeyValueResp{
-			KeyValue: &sdkws.KeyValue{
-				TypeKey:          model.TypeKey,
-				Value:            model.Value,
-				LatestUpdateTime: model.LatestUpdateTime,
-			},
-		}
-	}
-	return &pbMsg.ExtendMsg{
-		ReactionExtensions: reactionExtensionList,
-		ClientMsgID:        extendMsg.ClientMsgID,
-		MsgFirstModifyTime: extendMsg.MsgFirstModifyTime,
-		AttachedInfo:       extendMsg.AttachedInfo,
-		Ex:                 extendMsg.Ex,
-	}, nil
-}
-
-func (db *commonMsgDatabase) DeleteReactionExtendMsgSet(
-	ctx context.Context,
-	conversationID string,
-	sessionType int32,
-	clientMsgID string,
-	msgFirstModifyTime int64,
-	reactionExtensions map[string]*sdkws.KeyValue,
-) error {
-	return db.extendMsgDatabase.DeleteReactionExtendMsgSet(
-		ctx,
-		conversationID,
-		sessionType,
-		clientMsgID,
-		msgFirstModifyTime,
-		db.extendMsgSetModel.Pb2Model(reactionExtensions),
-	)
+func (db *commonMsgDatabase) RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error) {
+	return db.msgDocDatabase.RangeGroupSendCount(ctx, start, end, ase, pageNumber, showNumber)
 }

pkg/common/db/controller/s3.go

@@ -0,0 +1,76 @@
+package controller
+
+import "C"
+import (
+	"context"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/s3"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/s3/cont"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"path/filepath"
+	"time"
+)
+
+type S3Database interface {
+	PartLimit() *s3.PartLimit
+	PartSize(ctx context.Context, size int64) (int64, error)
+	AuthSign(ctx context.Context, uploadID string, partNumbers []int) (*s3.AuthSignResult, error)
+	InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int) (*cont.InitiateUploadResult, error)
+	CompleteMultipartUpload(ctx context.Context, uploadID string, parts []string) (*cont.UploadResult, error)
+	AccessURL(ctx context.Context, name string, expire time.Duration) (time.Time, string, error)
+	SetObject(ctx context.Context, info *relation.ObjectModel) error
+}
+
+func NewS3Database(s3 s3.Interface, obj relation.ObjectInfoModelInterface) S3Database {
+	return &s3Database{
+		s3:  cont.New(s3),
+		obj: obj,
+	}
+}
+
+type s3Database struct {
+	s3  *cont.Controller
+	obj relation.ObjectInfoModelInterface
+}
+
+func (s *s3Database) PartSize(ctx context.Context, size int64) (int64, error) {
+	return s.s3.PartSize(ctx, size)
+}
+
+func (s *s3Database) PartLimit() *s3.PartLimit {
+	return s.s3.PartLimit()
+}
+
+func (s *s3Database) AuthSign(ctx context.Context, uploadID string, partNumbers []int) (*s3.AuthSignResult, error) {
+	return s.s3.AuthSign(ctx, uploadID, partNumbers)
+}
+
+func (s *s3Database) InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int) (*cont.InitiateUploadResult, error) {
+	return s.s3.InitiateUpload(ctx, hash, size, expire, maxParts)
+}
+
+func (s *s3Database) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []string) (*cont.UploadResult, error) {
+	return s.s3.CompleteUpload(ctx, uploadID, parts)
+}
+
+func (s *s3Database) SetObject(ctx context.Context, info *relation.ObjectModel) error {
+	return s.obj.SetObject(ctx, info)
+}
+
+func (s *s3Database) AccessURL(ctx context.Context, name string, expire time.Duration) (time.Time, string, error) {
+	obj, err := s.obj.Take(ctx, name)
+	if err != nil {
+		return time.Time{}, "", err
+	}
+	opt := &s3.AccessURLOption{
+		ContentType: obj.ContentType,
+	}
+	if filename := filepath.Base(obj.Name); filename != "" {
+		opt.ContentDisposition = `attachment; filename=` + filename
+	}
+	expireTime := time.Now().Add(expire)
+	rawURL, err := s.s3.AccessURL(ctx, obj.Key, expire, opt)
+	if err != nil {
+		return time.Time{}, "", err
+	}
+	return expireTime, rawURL, nil
+}

pkg/common/db/controller/storage.go

@@ -1,567 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package controller
-
-import (
-	"bytes"
-	"context"
-	"crypto/md5"
-	"encoding/hex"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"net/url"
-	"path"
-	"strconv"
-	"time"
-
-	"github.com/google/uuid"
-
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/obj"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/third"
-	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
-)
-
-const (
-	hashPrefix     = "hash"
-	tempPrefix     = "temp"
-	fragmentPrefix = "fragment_"
-	urlsName       = "urls.json"
-)
-
-type S3Database interface {
-	ApplyPut(ctx context.Context, req *third.ApplyPutReq) (*third.ApplyPutResp, error)
-	GetPut(ctx context.Context, req *third.GetPutReq) (*third.GetPutResp, error)
-	ConfirmPut(ctx context.Context, req *third.ConfirmPutReq) (*third.ConfirmPutResp, error)
-	GetUrl(ctx context.Context, req *third.GetUrlReq) (*third.GetUrlResp, error)
-	GetHashInfo(ctx context.Context, req *third.GetHashInfoReq) (*third.GetHashInfoResp, error)
-	CleanExpirationObject(ctx context.Context, t time.Time)
-}
-
-func NewS3Database(
-	obj obj.Interface,
-	hash relation.ObjectHashModelInterface,
-	info relation.ObjectInfoModelInterface,
-	put relation.ObjectPutModelInterface,
-	url *url.URL,
-) S3Database {
-	return &s3Database{
-		url:  url,
-		obj:  obj,
-		hash: hash,
-		info: info,
-		put:  put,
-	}
-}
-
-type s3Database struct {
-	url  *url.URL
-	obj  obj.Interface
-	hash relation.ObjectHashModelInterface
-	info relation.ObjectInfoModelInterface
-	put  relation.ObjectPutModelInterface
-}
-
-// today 今天的日期
-func (c *s3Database) today() string {
-	return time.Now().Format("20060102")
-}
-
-// fragmentName 根据序号生成文件名
-func (c *s3Database) fragmentName(index int) string {
-	return fragmentPrefix + strconv.Itoa(index+1)
-}
-
-// getFragmentNum 获取分片大小和分片数量
-func (c *s3Database) getFragmentNum(fragmentSize int64, objectSize int64) (int64, int) {
-	if size := c.obj.MinFragmentSize(); fragmentSize < size {
-		fragmentSize = size
-	}
-	if fragmentSize <= 0 || objectSize <= fragmentSize {
-		return objectSize, 1
-	} else {
-		num := int(objectSize / fragmentSize)
-		if objectSize%fragmentSize > 0 {
-			num++
-		}
-		if n := c.obj.MaxFragmentNum(); num > n {
-			num = n
-		}
-		return fragmentSize, num
-	}
-}
-
-func (c *s3Database) CheckHash(hash string) error {
-	val, err := hex.DecodeString(hash)
-	if err != nil {
-		return err
-	}
-	if len(val) != md5.Size {
-		return errs.ErrArgs.Wrap("invalid hash")
-	}
-	return nil
-}
-
-func (c *s3Database) urlName(name string) string {
-	u := url.URL{
-		Scheme:      c.url.Scheme,
-		Opaque:      c.url.Opaque,
-		User:        c.url.User,
-		Host:        c.url.Host,
-		Path:        c.url.Path,
-		RawPath:     c.url.RawPath,
-		ForceQuery:  c.url.ForceQuery,
-		RawQuery:    c.url.RawQuery,
-		Fragment:    c.url.Fragment,
-		RawFragment: c.url.RawFragment,
-	}
-	v := make(url.Values, 1)
-	v.Set("name", name)
-	u.RawQuery = v.Encode()
-	return u.String()
-}
-
-func (c *s3Database) UUID() string {
-	return uuid.New().String()
-}
-
-func (c *s3Database) HashName(hash string) string {
-	return path.Join(hashPrefix, hash+"_"+c.today()+"_"+c.UUID())
-}
-
-func (c *s3Database) isNotFound(err error) bool {
-	return relation.IsNotFound(err)
-}
-
-func (c *s3Database) ApplyPut(ctx context.Context, req *third.ApplyPutReq) (*third.ApplyPutResp, error) {
-	if err := c.CheckHash(req.Hash); err != nil {
-		return nil, err
-	}
-	if err := c.obj.CheckName(req.Name); err != nil {
-		return nil, err
-	}
-	if req.ValidTime != 0 && req.ValidTime <= time.Now().UnixMilli() {
-		return nil, errors.New("invalid ValidTime")
-	}
-	var expirationTime *time.Time
-	if req.ValidTime != 0 {
-		expirationTime = utils.ToPtr(time.UnixMilli(req.ValidTime))
-	}
-	if hash, err := c.hash.Take(ctx, req.Hash, c.obj.Name()); err == nil {
-		o := relation.ObjectInfoModel{
-			Name:        req.Name,
-			Hash:        hash.Hash,
-			ValidTime:   expirationTime,
-			ContentType: req.ContentType,
-			CreateTime:  time.Now(),
-		}
-		if err := c.info.SetObject(ctx, &o); err != nil {
-			return nil, err
-		}
-		return &third.ApplyPutResp{Url: c.urlName(o.Name)}, nil // 服务器已存在
-	} else if !c.isNotFound(err) {
-		return nil, err
-	}
-	// 新上传
-	var fragmentNum int
-	const effective = time.Hour * 24 * 2
-	req.FragmentSize, fragmentNum = c.getFragmentNum(req.FragmentSize, req.Size)
-	put := relation.ObjectPutModel{
-		PutID:         req.PutID,
-		Hash:          req.Hash,
-		Name:          req.Name,
-		ObjectSize:    req.Size,
-		ContentType:   req.ContentType,
-		FragmentSize:  req.FragmentSize,
-		ValidTime:     expirationTime,
-		EffectiveTime: time.Now().Add(effective),
-	}
-	if put.PutID == "" {
-		put.PutID = c.UUID()
-	}
-	if v, err := c.put.Take(ctx, put.PutID); err == nil {
-		now := time.Now().UnixMilli()
-		if v.EffectiveTime.UnixMilli() <= now {
-			if err := c.put.DelPut(ctx, []string{v.PutID}); err != nil {
-				return nil, err
-			}
-		} else {
-			return nil, errs.ErrDuplicateKey.Wrap(fmt.Sprintf("duplicate put id %s", put.PutID))
-		}
-	} else if !c.isNotFound(err) {
-		return nil, err
-	}
-	put.Path = path.Join(tempPrefix, c.today(), req.Hash, put.PutID)
-	putURLs := make([]string, 0, fragmentNum)
-	for i := 0; i < fragmentNum; i++ {
-		url, err := c.obj.PresignedPutURL(ctx, &obj.ApplyPutArgs{
-			Bucket:        c.obj.TempBucket(),
-			Name:          path.Join(put.Path, c.fragmentName(i)),
-			Effective:     effective,
-			MaxObjectSize: req.FragmentSize,
-		})
-		if err != nil {
-			return nil, err
-		}
-		putURLs = append(putURLs, url)
-	}
-	urlsJsonData, err := json.Marshal(putURLs)
-	if err != nil {
-		return nil, err
-	}
-	t := md5.Sum(urlsJsonData)
-	put.PutURLsHash = hex.EncodeToString(t[:])
-	_, err = c.obj.PutObject(
-		ctx,
-		&obj.BucketObject{Bucket: c.obj.TempBucket(), Name: path.Join(put.Path, urlsName)},
-		bytes.NewReader(urlsJsonData),
-		int64(len(urlsJsonData)),
-	)
-	if err != nil {
-		return nil, err
-	}
-	put.CreateTime = time.Now()
-	if err := c.put.Create(ctx, []*relation.ObjectPutModel{&put}); err != nil {
-		return nil, err
-	}
-	return &third.ApplyPutResp{
-		PutID:        put.PutID,
-		FragmentSize: put.FragmentSize,
-		PutURLs:      putURLs,
-		ValidTime:    put.EffectiveTime.UnixMilli(),
-	}, nil
-}
-
-func (c *s3Database) GetPut(ctx context.Context, req *third.GetPutReq) (*third.GetPutResp, error) {
-	up, err := c.put.Take(ctx, req.PutID)
-	if err != nil {
-		return nil, err
-	}
-	reader, err := c.obj.GetObject(
-		ctx,
-		&obj.BucketObject{Bucket: c.obj.TempBucket(), Name: path.Join(up.Path, urlsName)},
-	)
-	if err != nil {
-		return nil, err
-	}
-	urlsData, err := io.ReadAll(reader)
-	if err != nil {
-		return nil, err
-	}
-	t := md5.Sum(urlsData)
-	if h := hex.EncodeToString(t[:]); h != up.PutURLsHash {
-		return nil, fmt.Errorf("invalid put urls hash %s %s", h, up.PutURLsHash)
-	}
-	var urls []string
-	if err := json.Unmarshal(urlsData, &urls); err != nil {
-		return nil, err
-	}
-	_, fragmentNum := c.getFragmentNum(up.FragmentSize, up.ObjectSize)
-	if len(urls) != fragmentNum {
-		return nil, fmt.Errorf("invalid urls length %d fragment %d", len(urls), fragmentNum)
-	}
-	fragments := make([]*third.GetPutFragment, fragmentNum)
-	for i := 0; i < fragmentNum; i++ {
-		name := path.Join(up.Path, c.fragmentName(i))
-		o, err := c.obj.GetObjectInfo(ctx, &obj.BucketObject{
-			Bucket: c.obj.TempBucket(),
-			Name:   name,
-		})
-		if err != nil {
-			if c.obj.IsNotFound(err) {
-				fragments[i] = &third.GetPutFragment{Url: urls[i]}
-				continue
-			}
-			return nil, err
-		}
-		fragments[i] = &third.GetPutFragment{Size: o.Size, Hash: o.Hash, Url: urls[i]}
-	}
-	var validTime int64
-	if up.ValidTime != nil {
-		validTime = up.ValidTime.UnixMilli()
-	}
-	return &third.GetPutResp{
-		FragmentSize: up.FragmentSize,
-		Size:         up.ObjectSize,
-		Name:         up.Name,
-		Hash:         up.Hash,
-		Fragments:    fragments,
-		PutURLsHash:  up.PutURLsHash,
-		ContentType:  up.ContentType,
-		ValidTime:    validTime,
-	}, nil
-}
-
-func (c *s3Database) ConfirmPut(ctx context.Context, req *third.ConfirmPutReq) (_ *third.ConfirmPutResp, _err error) {
-	put, err := c.put.Take(ctx, req.PutID)
-	if err != nil {
-		return nil, err
-	}
-	_, pack := c.getFragmentNum(put.FragmentSize, put.ObjectSize)
-	defer func() {
-		if _err == nil {
-			// 清理上传的碎片
-			err := c.obj.DeleteObject(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: put.Path})
-			if err != nil {
-				log.ZError(ctx, "deleteObject failed", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
-			}
-		}
-	}()
-	now := time.Now().UnixMilli()
-	if put.EffectiveTime.UnixMilli() < now {
-		return nil, errs.ErrFileUploadedExpired.Wrap("put expired")
-	}
-	if put.ValidTime != nil && put.ValidTime.UnixMilli() < now {
-		return nil, errs.ErrFileUploadedExpired.Wrap("object expired")
-	}
-	if hash, err := c.hash.Take(ctx, put.Hash, c.obj.Name()); err == nil {
-		o := relation.ObjectInfoModel{
-			Name:        put.Name,
-			Hash:        hash.Hash,
-			ValidTime:   put.ValidTime,
-			ContentType: put.ContentType,
-			CreateTime:  time.Now(),
-		}
-		if err := c.info.SetObject(ctx, &o); err != nil {
-			return nil, err
-		}
-		defer func() {
-			err := c.obj.DeleteObject(ctx, &obj.BucketObject{
-				Bucket: c.obj.TempBucket(),
-				Name:   put.Path,
-			})
-			if err != nil {
-				log.ZError(ctx, "DeleteObject", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
-			}
-		}()
-		// 服务端已存在
-		return &third.ConfirmPutResp{
-			Url: c.urlName(o.Name),
-		}, nil
-	} else if !c.isNotFound(err) {
-		return nil, err
-	}
-	src := make([]obj.BucketObject, pack)
-	for i := 0; i < pack; i++ {
-		name := path.Join(put.Path, c.fragmentName(i))
-		o, err := c.obj.GetObjectInfo(ctx, &obj.BucketObject{
-			Bucket: c.obj.TempBucket(),
-			Name:   name,
-		})
-		if err != nil {
-			return nil, err
-		}
-		if i+1 == pack { // 最后一个
-			size := put.ObjectSize - put.FragmentSize*int64(i)
-			if size != o.Size {
-				return nil, fmt.Errorf("last fragment %d size %d not equal to %d hash %s", i, o.Size, size, o.Hash)
-			}
-		} else {
-			if o.Size != put.FragmentSize {
-				return nil, fmt.Errorf("fragment %d size %d not equal to %d hash %s", i, o.Size, put.FragmentSize, o.Hash)
-			}
-		}
-		src[i] = obj.BucketObject{
-			Bucket: c.obj.TempBucket(),
-			Name:   name,
-		}
-	}
-	dst := &obj.BucketObject{
-		Bucket: c.obj.DataBucket(),
-		Name:   c.HashName(put.Hash),
-	}
-	if len(src) == 1 { // 未分片直接触发copy
-		// 检查数据完整性,避免脏数据
-		o, err := c.obj.GetObjectInfo(ctx, &src[0])
-		if err != nil {
-			return nil, err
-		}
-		if put.ObjectSize != o.Size {
-			return nil, fmt.Errorf("size mismatching should %d reality %d", put.ObjectSize, o.Size)
-		}
-		if put.Hash != o.Hash {
-			return nil, fmt.Errorf("hash mismatching should %s reality %s", put.Hash, o.Hash)
-		}
-		if err := c.obj.CopyObject(ctx, &src[0], dst); err != nil {
-			return nil, err
-		}
-	} else {
-		tempBucket := &obj.BucketObject{
-			Bucket: c.obj.TempBucket(),
-			Name:   path.Join(put.Path, "merge_"+c.UUID()),
-		}
-		defer func() { // 清理合成的文件
-			if err := c.obj.DeleteObject(ctx, tempBucket); err != nil {
-				log.ZError(ctx, "DeleteObject", err, "Bucket", tempBucket.Bucket, "Path", tempBucket.Name)
-			}
-		}()
-		err := c.obj.ComposeObject(ctx, src, tempBucket)
-		if err != nil {
-			return nil, err
-		}
-		info, err := c.obj.GetObjectInfo(ctx, tempBucket)
-		if err != nil {
-			return nil, err
-		}
-		if put.ObjectSize != info.Size {
-			return nil, fmt.Errorf("size mismatch should %d reality %d", put.ObjectSize, info.Size)
-		}
-		if put.Hash != info.Hash {
-			return nil, fmt.Errorf("hash mismatch should %s reality %s", put.Hash, info.Hash)
-		}
-		if err := c.obj.CopyObject(ctx, tempBucket, dst); err != nil {
-			return nil, err
-		}
-	}
-	h := &relation.ObjectHashModel{
-		Hash:       put.Hash,
-		Engine:     c.obj.Name(),
-		Size:       put.ObjectSize,
-		Bucket:     c.obj.DataBucket(),
-		Name:       dst.Name,
-		CreateTime: time.Now(),
-	}
-	if err := c.hash.Create(ctx, []*relation.ObjectHashModel{h}); err != nil {
-		return nil, err
-	}
-	o := &relation.ObjectInfoModel{
-		Name:        put.Name,
-		Hash:        put.Hash,
-		ContentType: put.ContentType,
-		ValidTime:   put.ValidTime,
-		CreateTime:  time.Now(),
-	}
-	if err := c.info.SetObject(ctx, o); err != nil {
-		return nil, err
-	}
-	if err := c.put.DelPut(ctx, []string{put.PutID}); err != nil {
-		log.ZError(ctx, "DelPut", err, "PutID", put.PutID)
-	}
-	return &third.ConfirmPutResp{
-		Url: c.urlName(o.Name),
-	}, nil
-}
-
-func (c *s3Database) GetUrl(ctx context.Context, req *third.GetUrlReq) (*third.GetUrlResp, error) {
-	info, err := c.info.Take(ctx, req.Name)
-	if err != nil {
-		return nil, err
-	}
-	if info.ValidTime != nil && info.ValidTime.Before(time.Now()) {
-		return nil, errs.ErrRecordNotFound.Wrap("object expired")
-	}
-	hash, err := c.hash.Take(ctx, info.Hash, c.obj.Name())
-	if err != nil {
-		return nil, err
-	}
-	opt := obj.HeaderOption{ContentType: info.ContentType}
-	if req.Attachment {
-		opt.Filename = info.Name
-	}
-	u, err := c.obj.PresignedGetURL(ctx, hash.Bucket, hash.Name, time.Duration(req.Expires)*time.Millisecond, &opt)
-	if err != nil {
-		return nil, err
-	}
-	return &third.GetUrlResp{
-		Url:  u,
-		Size: hash.Size,
-		Hash: hash.Hash,
-	}, nil
-}
-
-func (c *s3Database) CleanExpirationObject(ctx context.Context, t time.Time) {
-	// 清理上传产生的临时文件
-	c.cleanPutTemp(ctx, t, 10)
-	// 清理hash引用全过期的文件
-	c.cleanExpirationObject(ctx, t)
-	// 清理没有引用的hash对象
-	c.clearNoCitation(ctx, c.obj.Name(), 10)
-}
-
-func (c *s3Database) cleanPutTemp(ctx context.Context, t time.Time, num int) {
-	for {
-		puts, err := c.put.FindExpirationPut(ctx, t, num)
-		if err != nil {
-			log.ZError(ctx, "FindExpirationPut", err, "Time", t, "Num", num)
-			return
-		}
-		if len(puts) == 0 {
-			return
-		}
-		for _, put := range puts {
-			err := c.obj.DeleteObject(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: put.Path})
-			if err != nil {
-				log.ZError(ctx, "DeleteObject", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
-				return
-			}
-		}
-		ids := utils.Slice(puts, func(e *relation.ObjectPutModel) string { return e.PutID })
-		err = c.put.DelPut(ctx, ids)
-		if err != nil {
-			log.ZError(ctx, "DelPut", err, "PutID", ids)
-			return
-		}
-	}
-}
-
-func (c *s3Database) cleanExpirationObject(ctx context.Context, t time.Time) {
-	err := c.info.DeleteExpiration(ctx, t)
-	if err != nil {
-		log.ZError(ctx, "DeleteExpiration", err, "Time", t)
-	}
-}
-
-func (c *s3Database) clearNoCitation(ctx context.Context, engine string, limit int) {
-	for {
-		list, err := c.hash.DeleteNoCitation(ctx, engine, limit)
-		if err != nil {
-			log.ZError(ctx, "DeleteNoCitation", err, "Engine", engine, "Limit", limit)
-			return
-		}
-		if len(list) == 0 {
-			return
-		}
-		var hasErr bool
-		for _, h := range list {
-			err := c.obj.DeleteObject(ctx, &obj.BucketObject{Bucket: h.Bucket, Name: h.Name})
-			if err != nil {
-				hasErr = true
-				log.ZError(ctx, "DeleteObject", err, "Bucket", h.Bucket, "Path", h.Name)
-				continue
-			}
-		}
-		if hasErr {
-			return
-		}
-	}
-}
-
-func (c *s3Database) GetHashInfo(ctx context.Context, req *third.GetHashInfoReq) (*third.GetHashInfoResp, error) {
-	if err := c.CheckHash(req.Hash); err != nil {
-		return nil, err
-	}
-	o, err := c.hash.Take(ctx, req.Hash, c.obj.Name())
-	if err != nil {
-		return nil, err
-	}
-	return &third.GetHashInfoResp{
-		Hash: o.Hash,
-		Size: o.Size,
-	}, nil
-}

pkg/common/db/controller/user.go

@@ -45,7 +45,7 @@ type UserDatabase interface {
 	//函数内部先查询db中是否存在，存在则什么都不做；不存在则插入
 	InitOnce(ctx context.Context, users []*relation.UserModel) (err error)
 	// 获取用户总数
-	CountTotal(ctx context.Context) (int64, error)
+	CountTotal(ctx context.Context, before *time.Time) (int64, error)
 	// 获取范围内用户增量
 	CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error)
 }
@@ -151,8 +151,8 @@ func (u *userDatabase) GetAllUserID(ctx context.Context) (userIDs []string, err
 	return u.userDB.GetAllUserID(ctx)
 }
 
-func (u *userDatabase) CountTotal(ctx context.Context) (count int64, err error) {
-	return u.userDB.CountTotal(ctx)
+func (u *userDatabase) CountTotal(ctx context.Context, before *time.Time) (count int64, err error) {
+	return u.userDB.CountTotal(ctx, before)
 }
 
 func (u *userDatabase) CountRangeEverydayTotal(