feat: support GetLastMessage (#3029)

* pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * upgrading pkg tools * fix * fix * optimize log output * feat: support GetLastMessage * feat: support GetLastMessage * feat: s3 switch * feat: s3 switch
openimsdk · Jan 2, 2025 · ed0a834 · ed0a834
1 parent a1b5f05
commit ed0a834
Show file tree

Hide file tree

Showing 14 changed files with 367 additions and 4 deletions.
diff --git a/go.mod b/go.mod
@@ -12,7 +12,7 @@ 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/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

diff --git a/go.sum b/go.sum
@@ -347,8 +347,8 @@ 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/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=

diff --git a/internal/msggateway/client.go b/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:

diff --git a/internal/msggateway/constant.go b/internal/msggateway/constant.go
@@ -47,6 +47,7 @@ const (
 	WSSendSignalMsg       = 1004
 	WSPullMsg             = 1005
 	WSGetConvMaxReadSeq   = 1006
+	WsPullConvLastMessage = 1007
 	WSPushMsg             = 2001
 	WSKickOnlineMsg       = 2002
 	WsLogoutMsg           = 2003

diff --git a/internal/msggateway/message_handler.go b/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)
+}
diff --git a/internal/rpc/msg/sync_msg.go b/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
+}
diff --git a/pkg/common/storage/controller/msg.go b/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
 }
diff --git a/pkg/common/storage/database/mgo/msg.go b/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

diff --git a/pkg/common/storage/database/mgo/object.go b/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)
+}
diff --git a/pkg/common/storage/database/msg.go b/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)
 }
diff --git a/pkg/common/storage/database/object.go b/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
 }
diff --git a/tools/s3/README.md b/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
+```