Merge pull request #3032 from openimsdk/cherry-pick-a1b5f05

deps: Merge  #3026 #3029 PRs into pre-release-v3.8.3

config/openim-rpc-third.yml

@@ -38,3 +38,10 @@ object:
     accessKeySecret: 
     sessionToken: 
     publicRead: false
+  aws:
+    region: ap-southeast-2
+    bucket: testdemo832234
+    accessKeyID:
+    secretAccessKey:
+    sessionToken:
+    publicRead: false

go.mod

@@ -12,8 +12,8 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.72-alpha.69
-	github.com/openimsdk/tools v0.0.50-alpha.62
+	github.com/openimsdk/protocol v0.0.72-alpha.70
+	github.com/openimsdk/tools v0.0.50-alpha.63
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
 	github.com/stretchr/testify v1.9.0

go.sum

@@ -317,12 +317,12 @@ github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
-github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM=
-github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.72-alpha.69 h1:b22oY2XTdBR/BePqA73KsrM3GDF3Vk8YcBEXZU4ArJc=
-github.com/openimsdk/protocol v0.0.72-alpha.69/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
-github.com/openimsdk/tools v0.0.50-alpha.62 h1:e/m1XL7+EXbkOoxr/En/612WcOPKOUHPBj0++gG6MuQ=
-github.com/openimsdk/tools v0.0.50-alpha.62/go.mod h1:JowL2jYr8tu4vcQe+5hJh4v3BtSx1T0CIS3pgU/Mw+U=
+github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
+github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
+github.com/openimsdk/protocol v0.0.72-alpha.70 h1:j7vB81+rTthijRda2b8tlli9oWvPxr4yXHwZ8nPZIBQ=
+github.com/openimsdk/protocol v0.0.72-alpha.70/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
+github.com/openimsdk/tools v0.0.50-alpha.63 h1:dPoVvg4KWqYX/xtK3j96TwX2A/4jwT5S5XIHvSM9hTY=
+github.com/openimsdk/tools v0.0.50-alpha.63/go.mod h1:B+oqV0zdewN7OiEHYJm+hW+8/Te7B8tHHgD8rK5ZLZk=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

internal/msggateway/client.go

@@ -236,6 +236,8 @@ func (c *Client) handleMessage(message []byte) error {
 		resp, messageErr = c.longConnServer.GetSeqMessage(ctx, binaryReq)
 	case WSGetConvMaxReadSeq:
 		resp, messageErr = c.longConnServer.GetConversationsHasReadAndMaxSeq(ctx, binaryReq)
+	case WsPullConvLastMessage:
+		resp, messageErr = c.longConnServer.GetLastMessage(ctx, binaryReq)
 	case WsLogoutMsg:
 		resp, messageErr = c.longConnServer.UserLogout(ctx, binaryReq)
 	case WsSetBackgroundStatus:

internal/msggateway/constant.go

@@ -47,6 +47,7 @@ const (
 	WSSendSignalMsg       = 1004
 	WSPullMsg             = 1005
 	WSGetConvMaxReadSeq   = 1006
+	WsPullConvLastMessage = 1007
 	WSPushMsg             = 2001
 	WSKickOnlineMsg       = 2002
 	WsLogoutMsg           = 2003

internal/msggateway/hub_server.go

@@ -89,10 +89,6 @@ func NewServer(longConnServer LongConnServer, conf *Config, ready func(srv *Serv
 	return s
 }
 
-func (s *Server) OnlinePushMsg(context context.Context, req *msggateway.OnlinePushMsgReq) (*msggateway.OnlinePushMsgResp, error) {
-	panic("implement me")
-}
-
 func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUsersOnlineStatusReq) (*msggateway.GetUsersOnlineStatusResp, error) {
 	if !authverify.IsAppManagerUid(ctx, s.config.Share.IMAdminUserID) {
 		return nil, errs.ErrNoPermission.WrapMsg("only app manager")
@@ -126,11 +122,6 @@ func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUs
 	return &resp, nil
 }
 
-func (s *Server) OnlineBatchPushOneMsg(ctx context.Context, req *msggateway.OnlineBatchPushOneMsgReq) (*msggateway.OnlineBatchPushOneMsgResp, error) {
-	// todo implement
-	return nil, nil
-}
-
 func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.MsgData) *msggateway.SingleMsgToUserResults {
 	clients, ok := s.LongConnServer.GetUserAllCons(userID)
 	if !ok {

internal/msggateway/message_handler.go

@@ -108,6 +108,7 @@ type MessageHandler interface {
 	GetSeqMessage(ctx context.Context, data *Req) ([]byte, error)
 	UserLogout(ctx context.Context, data *Req) ([]byte, error)
 	SetUserDeviceBackground(ctx context.Context, data *Req) ([]byte, bool, error)
+	GetLastMessage(ctx context.Context, data *Req) ([]byte, error)
 }
 
 var _ MessageHandler = (*GrpcHandler)(nil)
@@ -266,3 +267,15 @@ func (g *GrpcHandler) SetUserDeviceBackground(ctx context.Context, data *Req) ([
 	}
 	return nil, req.IsBackground, nil
 }
+
+func (g *GrpcHandler) GetLastMessage(ctx context.Context, data *Req) ([]byte, error) {
+	var req msg.GetLastMessageReq
+	if err := proto.Unmarshal(data.Data, &req); err != nil {
+		return nil, err
+	}
+	resp, err := g.msgClient.GetLastMessage(ctx, &req)
+	if err != nil {
+		return nil, err
+	}
+	return proto.Marshal(resp)
+}

internal/push/offlinepush/dummy/push.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
 	"github.com/openimsdk/tools/log"
+	"sync/atomic"
 )
 
 func NewClient() *Dummy {
@@ -25,10 +26,12 @@ func NewClient() *Dummy {
 }
 
 type Dummy struct {
+	v atomic.Bool
 }
 
 func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
-	log.ZDebug(ctx, "dummy push")
-	log.ZWarn(ctx, "Dummy push", nil, "ps", "The offline push is not configured. To configure it, please go to config/openim-push.yml.")
+	if d.v.CompareAndSwap(false, true) {
+		log.ZWarn(ctx, "dummy push", nil, "ps", "the offline push is not configured. to configure it, please go to config/openim-push.yml")
+	}
 	return nil
 }

internal/rpc/msg/sync_msg.go

@@ -245,3 +245,11 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
 func (m *msgServer) GetServerTime(ctx context.Context, _ *msg.GetServerTimeReq) (*msg.GetServerTimeResp, error) {
 	return &msg.GetServerTimeResp{ServerTime: timeutil.GetCurrentTimestampByMill()}, nil
 }
+
+func (m *msgServer) GetLastMessage(ctx context.Context, req *msg.GetLastMessageReq) (*msg.GetLastMessageResp, error) {
+	msgs, err := m.MsgDatabase.GetLastMessage(ctx, req.ConversationIDs, req.UserID)
+	if err != nil {
+		return nil, err
+	}
+	return &msg.GetLastMessageResp{Msgs: msgs}, nil
+}

pkg/common/config/config.go

@@ -15,6 +15,7 @@
 package config
 
 import (
+	"github.com/openimsdk/tools/s3/aws"
 	"strings"
 	"time"
 
@@ -290,14 +291,7 @@ type Third struct {
 		Cos    Cos    `mapstructure:"cos"`
 		Oss    Oss    `mapstructure:"oss"`
 		Kodo   Kodo   `mapstructure:"kodo"`
-		Aws    struct {
-			Endpoint        string `mapstructure:"endpoint"`
-			Region          string `mapstructure:"region"`
-			Bucket          string `mapstructure:"bucket"`
-			AccessKeyID     string `mapstructure:"accessKeyID"`
-			AccessKeySecret string `mapstructure:"accessKeySecret"`
-			PublicRead      bool   `mapstructure:"publicRead"`
-		} `mapstructure:"aws"`
+		Aws    Aws    `mapstructure:"aws"`
 	} `mapstructure:"object"`
 }
 type Cos struct {
@@ -327,6 +321,15 @@ type Kodo struct {
 	PublicRead      bool   `mapstructure:"publicRead"`
 }
 
+type Aws struct {
+	Endpoint        string `mapstructure:"endpoint"`
+	Region          string `mapstructure:"region"`
+	Bucket          string `mapstructure:"bucket"`
+	AccessKeyID     string `mapstructure:"accessKeyID"`
+	SecretAccessKey string `mapstructure:"secretAccessKey"`
+	SessionToken    string `mapstructure:"sessionToken"`
+}
+
 type User struct {
 	RPC struct {
 		RegisterIP string `mapstructure:"registerIP"`
@@ -567,6 +570,16 @@ func (o *Kodo) Build() *kodo.Config {
 	}
 }
 
+func (o *Aws) Build() *aws.Config {
+	return &aws.Config{
+		Region:          o.Region,
+		Bucket:          o.Bucket,
+		AccessKeyID:     o.AccessKeyID,
+		SecretAccessKey: o.SecretAccessKey,
+		SessionToken:    o.SessionToken,
+	}
+}
+
 func (l *CacheConfig) Failed() time.Duration {
 	return time.Second * time.Duration(l.FailedExpire)
 }
@@ -578,3 +591,115 @@ func (l *CacheConfig) Success() time.Duration {
 func (l *CacheConfig) Enable() bool {
 	return l.Topic != "" && l.SlotNum > 0 && l.SlotSize > 0
 }
+
+var (
+	DiscoveryConfigFilename          = "discovery.yml"
+	KafkaConfigFileName              = "kafka.yml"
+	LocalCacheConfigFileName         = "local-cache.yml"
+	LogConfigFileName                = "log.yml"
+	MinioConfigFileName              = "minio.yml"
+	MongodbConfigFileName            = "mongodb.yml"
+	OpenIMAPICfgFileName             = "openim-api.yml"
+	OpenIMCronTaskCfgFileName        = "openim-crontask.yml"
+	OpenIMMsgGatewayCfgFileName      = "openim-msggateway.yml"
+	OpenIMMsgTransferCfgFileName     = "openim-msgtransfer.yml"
+	OpenIMPushCfgFileName            = "openim-push.yml"
+	OpenIMRPCAuthCfgFileName         = "openim-rpc-auth.yml"
+	OpenIMRPCConversationCfgFileName = "openim-rpc-conversation.yml"
+	OpenIMRPCFriendCfgFileName       = "openim-rpc-friend.yml"
+	OpenIMRPCGroupCfgFileName        = "openim-rpc-group.yml"
+	OpenIMRPCMsgCfgFileName          = "openim-rpc-msg.yml"
+	OpenIMRPCThirdCfgFileName        = "openim-rpc-third.yml"
+	OpenIMRPCUserCfgFileName         = "openim-rpc-user.yml"
+	RedisConfigFileName              = "redis.yml"
+	ShareFileName                    = "share.yml"
+	WebhooksConfigFileName           = "webhooks.yml"
+)
+
+func (d *Discovery) GetConfigFileName() string {
+	return DiscoveryConfigFilename
+}
+
+func (k *Kafka) GetConfigFileName() string {
+	return KafkaConfigFileName
+}
+
+func (lc *LocalCache) GetConfigFileName() string {
+	return LocalCacheConfigFileName
+}
+
+func (l *Log) GetConfigFileName() string {
+	return LogConfigFileName
+}
+
+func (m *Minio) GetConfigFileName() string {
+	return MinioConfigFileName
+}
+
+func (m *Mongo) GetConfigFileName() string {
+	return MongodbConfigFileName
+}
+
+func (n *Notification) GetConfigFileName() string {
+	return NotificationFileName
+}
+
+func (a *API) GetConfigFileName() string {
+	return OpenIMAPICfgFileName
+}
+
+func (ct *CronTask) GetConfigFileName() string {
+	return OpenIMCronTaskCfgFileName
+}
+
+func (mg *MsgGateway) GetConfigFileName() string {
+	return OpenIMMsgGatewayCfgFileName
+}
+
+func (mt *MsgTransfer) GetConfigFileName() string {
+	return OpenIMMsgTransferCfgFileName
+}
+
+func (p *Push) GetConfigFileName() string {
+	return OpenIMPushCfgFileName
+}
+
+func (a *Auth) GetConfigFileName() string {
+	return OpenIMRPCAuthCfgFileName
+}
+
+func (c *Conversation) GetConfigFileName() string {
+	return OpenIMRPCConversationCfgFileName
+}
+
+func (f *Friend) GetConfigFileName() string {
+	return OpenIMRPCFriendCfgFileName
+}
+
+func (g *Group) GetConfigFileName() string {
+	return OpenIMRPCGroupCfgFileName
+}
+
+func (m *Msg) GetConfigFileName() string {
+	return OpenIMRPCMsgCfgFileName
+}
+
+func (t *Third) GetConfigFileName() string {
+	return OpenIMRPCThirdCfgFileName
+}
+
+func (u *User) GetConfigFileName() string {
+	return OpenIMRPCUserCfgFileName
+}
+
+func (r *Redis) GetConfigFileName() string {
+	return RedisConfigFileName
+}
+
+func (s *Share) GetConfigFileName() string {
+	return ShareFileName
+}
+
+func (w *Webhooks) GetConfigFileName() string {
+	return WebhooksConfigFileName
+}

pkg/common/storage/controller/msg.go

@@ -97,6 +97,8 @@ type CommonMsgDatabase interface {
 	DeleteDoc(ctx context.Context, docID string) error
 
 	GetLastMessageSeqByTime(ctx context.Context, conversationID string, time int64) (int64, error)
+
+	GetLastMessage(ctx context.Context, conversationIDS []string, userID string) (map[string]*sdkws.MsgData, error)
 }
 
 func NewCommonMsgDatabase(msgDocModel database.Msg, msg cache.MsgCache, seqUser cache.SeqUser, seqConversation cache.SeqConversationCache, kafkaConf *config.Kafka) (CommonMsgDatabase, error) {
@@ -811,8 +813,29 @@ func (db *commonMsgDatabase) GetMessageBySeqs(ctx context.Context, conversationI
 		if v, ok := seqMsgs[seq]; ok {
 			res = append(res, convert.MsgDB2Pb(v.Msg))
 		} else {
-			res = append(res, &sdkws.MsgData{Seq: seq})
+			res = append(res, &sdkws.MsgData{Seq: seq, Status: constant.MsgStatusHasDeleted})
 		}
 	}
 	return res, nil
 }
+
+func (db *commonMsgDatabase) GetLastMessage(ctx context.Context, conversationIDs []string, userID string) (map[string]*sdkws.MsgData, error) {
+	res := make(map[string]*sdkws.MsgData)
+	for _, conversationID := range conversationIDs {
+		if _, ok := res[conversationID]; ok {
+			continue
+		}
+		msg, err := db.msgDocDatabase.GetLastMessage(ctx, conversationID)
+		if err != nil {
+			if errs.Unwrap(err) == mongo.ErrNoDocuments {
+				continue
+			}
+			return nil, err
+		}
+		tmp := []*model.MsgInfoModel{msg}
+		db.handlerDeleteAndRevoked(ctx, userID, tmp)
+		db.handlerQuote(ctx, userID, conversationID, tmp)
+		res[conversationID] = convert.MsgDB2Pb(msg.Msg)
+	}
+	return res, nil
+}

pkg/common/storage/database/mgo/msg.go

@@ -997,6 +997,68 @@ func (m *MsgMgo) GetLastMessageSeqByTime(ctx context.Context, conversationID str
 	return seq, nil
 }
 
+func (m *MsgMgo) GetLastMessage(ctx context.Context, conversationID string) (*model.MsgInfoModel, error) {
+	pipeline := []bson.M{
+		{
+			"$match": bson.M{
+				"doc_id": bson.M{
+					"$regex": fmt.Sprintf("^%s", conversationID),
+				},
+			},
+		},
+		{
+			"$match": bson.M{
+				"msgs.msg.status": bson.M{
+					"$lt": constant.MsgStatusHasDeleted,
+				},
+			},
+		},
+		{
+			"$sort": bson.M{
+				"_id": -1,
+			},
+		},
+		{
+			"$limit": 1,
+		},
+		{
+			"$project": bson.M{
+				"_id":    0,
+				"doc_id": 0,
+			},
+		},
+		{
+			"$unwind": "$msgs",
+		},
+		{
+			"$match": bson.M{
+				"msgs.msg.status": bson.M{
+					"$lt": constant.MsgStatusHasDeleted,
+				},
+			},
+		},
+		{
+			"$sort": bson.M{
+				"msgs.msg.seq": -1,
+			},
+		},
+		{
+			"$limit": 1,
+		},
+	}
+	type Result struct {
+		Msgs *model.MsgInfoModel `bson:"msgs"`
+	}
+	res, err := mongoutil.Aggregate[*Result](ctx, m.coll, pipeline)
+	if err != nil {
+		return nil, err
+	}
+	if len(res) == 0 {
+		return nil, errs.Wrap(mongo.ErrNoDocuments)
+	}
+	return res[0].Msgs, nil
+}
+
 func (m *MsgMgo) onlyFindDocIndex(ctx context.Context, docID string, indexes []int64) ([]*model.MsgInfoModel, error) {
 	if len(indexes) == 0 {
 		return nil, nil

pkg/common/storage/database/mgo/object.go

@@ -112,3 +112,15 @@ func (o *S3Mongo) FindExpirationObject(ctx context.Context, engine string, expir
 func (o *S3Mongo) GetKeyCount(ctx context.Context, engine string, key string) (int64, error) {
 	return mongoutil.Count(ctx, o.coll, bson.M{"engine": engine, "key": key})
 }
+
+func (o *S3Mongo) GetEngineCount(ctx context.Context, engine string) (int64, error) {
+	return mongoutil.Count(ctx, o.coll, bson.M{"engine": engine})
+}
+
+func (o *S3Mongo) GetEngineInfo(ctx context.Context, engine string, limit int, skip int) ([]*model.Object, error) {
+	return mongoutil.Find[*model.Object](ctx, o.coll, bson.M{"engine": engine}, options.Find().SetLimit(int64(limit)).SetSkip(int64(skip)))
+}
+
+func (o *S3Mongo) UpdateEngine(ctx context.Context, oldEngine, oldName string, newEngine string) error {
+	return mongoutil.UpdateOne(ctx, o.coll, bson.M{"engine": oldEngine, "name": oldName}, bson.M{"$set": bson.M{"engine": newEngine}}, false)
+}

pkg/common/storage/database/msg.go

@@ -39,5 +39,6 @@ type Msg interface {
 	DeleteDoc(ctx context.Context, docID string) error
 	GetRandBeforeMsg(ctx context.Context, ts int64, limit int) ([]*model.MsgDocModel, error)
 	GetLastMessageSeqByTime(ctx context.Context, conversationID string, time int64) (int64, error)
+	GetLastMessage(ctx context.Context, conversationID string) (*model.MsgInfoModel, error)
 	FindSeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error)
 }

pkg/common/storage/database/object.go

@@ -27,4 +27,8 @@ type ObjectInfo interface {
 	Delete(ctx context.Context, engine string, name []string) error
 	FindExpirationObject(ctx context.Context, engine string, expiration time.Time, needDelType []string, count int64) ([]*model.Object, error)
 	GetKeyCount(ctx context.Context, engine string, key string) (int64, error)
+
+	GetEngineCount(ctx context.Context, engine string) (int64, error)
+	GetEngineInfo(ctx context.Context, engine string, limit int, skip int) ([]*model.Object, error)
+	UpdateEngine(ctx context.Context, oldEngine, oldName string, newEngine string) error
 }

tools/s3/README.md

@@ -0,0 +1,12 @@
+# After s3 switches the storage engine, convert the data
+
+- build
+```shell
+go build -o s3convert main.go
+```
+
+- start
+```shell
+./s3convert -config <config dir path> -name <old s3 name>
+# ./s3convert -config ./../../config -name minio
+```

tools/s3/internal/conversion.go

@@ -0,0 +1,202 @@
+package internal
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/mitchellh/mapstructure"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
+	"github.com/openimsdk/tools/db/mongoutil"
+	"github.com/openimsdk/tools/db/redisutil"
+	"github.com/openimsdk/tools/s3"
+	"github.com/openimsdk/tools/s3/aws"
+	"github.com/openimsdk/tools/s3/cos"
+	"github.com/openimsdk/tools/s3/kodo"
+	"github.com/openimsdk/tools/s3/minio"
+	"github.com/openimsdk/tools/s3/oss"
+	"github.com/spf13/viper"
+	"go.mongodb.org/mongo-driver/mongo"
+	"log"
+	"net/http"
+	"path/filepath"
+	"time"
+)
+
+const defaultTimeout = time.Second * 10
+
+func readConf(path string, val any) error {
+	v := viper.New()
+	v.SetConfigFile(path)
+	if err := v.ReadInConfig(); err != nil {
+		return err
+	}
+	fn := func(config *mapstructure.DecoderConfig) {
+		config.TagName = "mapstructure"
+	}
+	return v.Unmarshal(val, fn)
+}
+
+func getS3(path string, name string, thirdConf *config.Third) (s3.Interface, error) {
+	switch name {
+	case "minio":
+		ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+		defer cancel()
+		var minioConf config.Minio
+		if err := readConf(filepath.Join(path, minioConf.GetConfigFileName()), &minioConf); err != nil {
+			return nil, err
+		}
+		var redisConf config.Redis
+		if err := readConf(filepath.Join(path, redisConf.GetConfigFileName()), &redisConf); err != nil {
+			return nil, err
+		}
+		rdb, err := redisutil.NewRedisClient(ctx, redisConf.Build())
+		if err != nil {
+			return nil, err
+		}
+		return minio.NewMinio(ctx, redis.NewMinioCache(rdb), *minioConf.Build())
+	case "cos":
+		return cos.NewCos(*thirdConf.Object.Cos.Build())
+	case "oss":
+		return oss.NewOSS(*thirdConf.Object.Oss.Build())
+	case "kodo":
+		return kodo.NewKodo(*thirdConf.Object.Kodo.Build())
+	case "aws":
+		return aws.NewAws(*thirdConf.Object.Aws.Build())
+	default:
+		return nil, fmt.Errorf("invalid object enable: %s", name)
+	}
+}
+
+func getMongo(path string) (database.ObjectInfo, error) {
+	var mongoConf config.Mongo
+	if err := readConf(filepath.Join(path, mongoConf.GetConfigFileName()), &mongoConf); err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	mgocli, err := mongoutil.NewMongoDB(ctx, mongoConf.Build())
+	if err != nil {
+		return nil, err
+	}
+	return mgo.NewS3Mongo(mgocli.GetDB())
+}
+
+func Main(path string, engine string) error {
+	var thirdConf config.Third
+	if err := readConf(filepath.Join(path, thirdConf.GetConfigFileName()), &thirdConf); err != nil {
+		return err
+	}
+	if thirdConf.Object.Enable == engine {
+		return errors.New("same s3 storage")
+	}
+	s3db, err := getMongo(path)
+	if err != nil {
+		return err
+	}
+	oldS3, err := getS3(path, engine, &thirdConf)
+	if err != nil {
+		return err
+	}
+	newS3, err := getS3(path, thirdConf.Object.Enable, &thirdConf)
+	if err != nil {
+		return err
+	}
+	count, err := getEngineCount(s3db, oldS3.Engine())
+	if err != nil {
+		return err
+	}
+	log.Printf("engine %s count: %d", oldS3.Engine(), count)
+	var skip int
+	for i := 1; i <= count+1; i++ {
+		log.Printf("start %d/%d", i, count)
+		start := time.Now()
+		res, err := doObject(s3db, newS3, oldS3, skip)
+		if err != nil {
+			log.Printf("end [%s] %d/%d error %s", time.Since(start), i, count, err)
+			return err
+		}
+		log.Printf("end [%s] %d/%d result %+v", time.Since(start), i, count, *res)
+		if res.Skip {
+			skip++
+		}
+		if res.End {
+			break
+		}
+	}
+	return nil
+}
+
+func getEngineCount(db database.ObjectInfo, name string) (int, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	count, err := db.GetEngineCount(ctx, name)
+	if err != nil {
+		return 0, err
+	}
+	return int(count), nil
+}
+
+func doObject(db database.ObjectInfo, newS3, oldS3 s3.Interface, skip int) (*Result, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	infos, err := db.GetEngineInfo(ctx, oldS3.Engine(), 1, skip)
+	if err != nil {
+		return nil, err
+	}
+	if len(infos) == 0 {
+		return &Result{End: true}, nil
+	}
+	obj := infos[0]
+	if _, err := db.Take(ctx, newS3.Engine(), obj.Name); err == nil {
+		return &Result{Skip: true}, nil
+	} else if !errors.Is(err, mongo.ErrNoDocuments) {
+		return nil, err
+	}
+	downloadURL, err := oldS3.AccessURL(ctx, obj.Key, time.Hour, &s3.AccessURLOption{})
+	if err != nil {
+		return nil, err
+	}
+	putURL, err := newS3.PresignedPutObject(ctx, obj.Key, time.Hour)
+	if err != nil {
+		return nil, err
+	}
+	downloadResp, err := http.Get(downloadURL)
+	if err != nil {
+		return nil, err
+	}
+	defer downloadResp.Body.Close()
+	switch downloadResp.StatusCode {
+	case http.StatusNotFound:
+		return &Result{Skip: true}, nil
+	case http.StatusOK:
+	default:
+		return nil, fmt.Errorf("download object failed %s", downloadResp.Status)
+	}
+	log.Printf("file size %d", obj.Size)
+	request, err := http.NewRequest(http.MethodPut, putURL, downloadResp.Body)
+	if err != nil {
+		return nil, err
+	}
+	putResp, err := http.DefaultClient.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer putResp.Body.Close()
+	if putResp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("put object failed %s", putResp.Status)
+	}
+	ctx, cancel = context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
+	if err := db.UpdateEngine(ctx, obj.Engine, obj.Name, newS3.Engine()); err != nil {
+		return nil, err
+	}
+	return &Result{}, nil
+}
+
+type Result struct {
+	Skip bool
+	End  bool
+}

tools/s3/main.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"github.com/openimsdk/open-im-server/v3/tools/s3/internal"
+	"os"
+)
+
+func main() {
+	var (
+		name   string
+		config string
+	)
+	flag.StringVar(&name, "name", "", "old previous storage name")
+	flag.StringVar(&config, "config", "", "config directory")
+	flag.Parse()
+	if err := internal.Main(config, name); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	fmt.Fprintln(os.Stdout, "success")
+}