Merge remote-tracking branch 'origin/errcode' into errcode

internal/api/msg.go

@@ -119,18 +119,6 @@ func (m *Message) PullMsgBySeqs(c *gin.Context) {
 	a2r.Call(msg.MsgClient.PullMessageBySeqs, m.client, c)
 }
 
-func (m *Message) DelMsg(c *gin.Context) {
-	a2r.Call(msg.MsgClient.DelMsgs, m.client, c)
-}
-
-func (m *Message) DelSuperGroupMsg(c *gin.Context) {
-	a2r.Call(msg.MsgClient.DelSuperGroupMsg, m.client, c)
-}
-
-func (m *Message) ClearMsg(c *gin.Context) {
-	a2r.Call(msg.MsgClient.ClearMsg, m.client, c)
-}
-
 func (m *Message) RevokeMsg(c *gin.Context) {
 	a2r.Call(msg.MsgClient.RevokeMsg, m.client, c)
 }

internal/api/route.go

@@ -139,9 +139,7 @@ func NewGinRouter(zk discoveryregistry.SvcDiscoveryRegistry, rdb redis.Universal
 		msgGroup.POST("/newest_seq", m.GetSeq)
 		msgGroup.POST("/send_msg", m.SendMessage)
 		msgGroup.POST("/pull_msg_by_seq", m.PullMsgBySeqs)
-		msgGroup.POST("/del_msg", m.DelMsg)
+		// todo del msg route
-		msgGroup.POST("/del_super_group_msg", m.DelSuperGroupMsg)
-		msgGroup.POST("/clear_msg", m.ClearMsg)
 		msgGroup.POST("/revoke_msg", m.RevokeMsg)
 
 		msgGroup.POST("/batch_send_msg", m.ManagementBatchSendMsg)

internal/rpc/msg/delete.go

@@ -3,30 +3,34 @@ package msg
 import (
 	"context"
 
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/tokenverify"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
 )
 
-func (m *msgServer) DelMsgs(ctx context.Context, req *msg.DelMsgsReq) (*msg.DelMsgsResp, error) {
+func (m *msgServer) getMinSeqs(maxSeqs map[string]int64) map[string]int64 {
-	// if _, err := m.MsgDatabase.DelMsgBySeqs(ctx, req.UserID, req.Seqs); err != nil {
+	minSeqs := make(map[string]int64)
-	// 	return nil, err
+	for k, v := range maxSeqs {
-	// }
+		minSeqs[k] = v + 1
-	return &msg.DelMsgsResp{}, nil
+	}
+	return minSeqs
 }
 
-func (m *msgServer) DelSuperGroupMsg(ctx context.Context, req *msg.DelSuperGroupMsgReq) (*msg.DelSuperGroupMsgResp, error) {
+func (m *msgServer) ClearConversationsMsg(ctx context.Context, req *msg.ClearConversationsMsgReq) (*msg.ClearConversationsMsgResp, error) {
-	resp := &msg.DelSuperGroupMsgResp{}
+	if err := tokenverify.CheckAccessV3(ctx, req.UserID); err != nil {
-	if err := tokenverify.CheckAdmin(ctx); err != nil {
 		return nil, err
 	}
-	if err := m.MsgDatabase.DeleteConversationMsgsAndSetMinSeq(ctx, req.GroupID, 0); err != nil {
+	maxSeqs, err := m.MsgDatabase.GetMaxSeqs(ctx, req.ConversationIDs)
+	if err != nil {
 		return nil, err
 	}
-	return resp, nil
+	if err := m.MsgDatabase.SetUserConversationsMinSeqs(ctx, req.UserID, m.getMinSeqs(maxSeqs)); err != nil {
+		return nil, err
+	}
+	return &msg.ClearConversationsMsgResp{}, nil
 }
 
-func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (*msg.ClearMsgResp, error) {
+func (m *msgServer) UserClearAllMsg(ctx context.Context, req *msg.UserClearAllMsgReq) (*msg.UserClearAllMsgResp, error) {
-	resp := &msg.ClearMsgResp{}
 	if err := tokenverify.CheckAccessV3(ctx, req.UserID); err != nil {
 		return nil, err
 	}
@@ -34,6 +38,32 @@ func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (*msg.Cl
 	if err != nil {
 		return nil, err
 	}
-	m.MsgDatabase.CleanUpUserConversationsMsgs(ctx, req.UserID, conversationIDs)
+	maxSeqs, err := m.MsgDatabase.GetMaxSeqs(ctx, conversationIDs)
-	return resp, nil
+	if err != nil {
+		return nil, err
+	}
+	if err := m.MsgDatabase.SetUserConversationsMinSeqs(ctx, req.UserID, m.getMinSeqs(maxSeqs)); err != nil {
+		return nil, err
+	}
+	return &msg.UserClearAllMsgResp{}, nil
+}
+
+func (m *msgServer) DeleteMsgs(ctx context.Context, req *msg.DeleteMsgsReq) (*msg.DeleteMsgsResp, error) {
+	if err := tokenverify.CheckAccessV3(ctx, req.UserID); err != nil {
+		return nil, err
+	}
+	return &msg.DeleteMsgsResp{}, nil
+}
+
+func (m *msgServer) DeleteMsgPhysicalBySeq(ctx context.Context, req *msg.DeleteMsgPhysicalBySeqReq) (*msg.DeleteMsgPhysicalBySeqResp, error) {
+	return &msg.DeleteMsgPhysicalBySeqResp{}, nil
+}
+
+func (m *msgServer) DeleteMsgPhysical(ctx context.Context, req *msg.DeleteMsgPhysicalReq) (*msg.DeleteMsgPhysicalResp, error) {
+	for _, conversationID := range req.ConversationIDs {
+		if err := m.MsgDatabase.DeleteConversationMsgsAndSetMinSeq(ctx, conversationID, req.RemainTime); err != nil {
+			log.ZWarn(ctx, "DeleteConversationMsgsAndSetMinSeq error", err, "conversationID", conversationID, "err", err)
+		}
+	}
+	return &msg.DeleteMsgPhysicalResp{}, nil
 }

pkg/common/db/cache/msg.go

@@ -52,7 +52,10 @@ type MsgModel interface {
 	GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error)
 	GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error)
 	SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error
+	// seqs map: key userID value minSeq
 	SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error)
+	// seqs map: key conversationID value minSeq
+	SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) error
 
 	AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error
 	GetTokensWithoutError(ctx context.Context, userID, platformID string) (map[string]int, error)
@@ -195,6 +198,18 @@ func (c *msgCache) SetConversationUserMinSeqs(ctx context.Context, conversationI
 	return err
 }
 
+func (c *msgCache) SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) (err error) {
+	pipe := c.rdb.Pipeline()
+	for conversationID, minSeq := range seqs {
+		err = pipe.Set(ctx, c.getConversationUserMinSeqKey(conversationID, userID), minSeq, 0).Err()
+		if err != nil {
+			return errs.Wrap(err)
+		}
+	}
+	_, err = pipe.Exec(ctx)
+	return err
+}
+
 func (c *msgCache) AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
 	key := uidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
 	return errs.Wrap(c.rdb.HSet(ctx, key, token, flag).Err())

pkg/common/db/controller/msg.go

@@ -58,6 +58,7 @@ type CommonMsgDatabase interface {
 	GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error)
 	SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error
 	SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error)
+	SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) (err 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)
@@ -756,6 +757,10 @@ func (db *commonMsgDatabase) SetConversationUserMinSeqs(ctx context.Context, con
 	return db.cache.SetConversationUserMinSeqs(ctx, conversationID, seqs)
 }
 
+func (db *commonMsgDatabase) SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) error {
+	return db.cache.SetUserConversationsMinSeqs(ctx, userID, seqs)
+}
+
 func (db *commonMsgDatabase) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
 	return db.cache.SetSendMsgStatus(ctx, id, status)
 }

pkg/proto/msg/msg.proto

@@ -44,21 +44,6 @@ message SendMsgResp {
 }
 
 
-message ClearMsgReq{
-  string userID = 1;
-}
-
-
-message ClearMsgResp{
-}
-
-message SetMsgMinSeqReq{
-  string userID = 1;
-  string groupID = 2;
-  uint32 minSeq = 3;
-}
-message SetMsgMinSeqResp{
-}
 
 message SetSendMsgStatusReq{
   int32 status = 1;
@@ -74,28 +59,6 @@ message GetSendMsgStatusResp{
   int32 status = 1;
 }
 
-message DelSuperGroupMsgReq{
-  string userID = 1;
-  string groupID = 2;
-}
-
-message DelSuperGroupMsgResp{
-}
-
-message GetSuperGroupMsgReq{
-  int64 Seq = 1;
-  string groupID = 2;
-}
-message GetSuperGroupMsgResp{
-  sdkws.MsgData msgData = 1;
-}
-
-message GetWriteDiffMsgReq{
-  int64 Seq = 1;
-}
-message GetWriteDiffMsgResp{
-  sdkws.MsgData msgData = 2;
-}
 
 message ModifyMessageReactionExtensionsReq {
   string conversationID = 1;
@@ -205,6 +168,46 @@ message RevokeMsgReq {
 message RevokeMsgResp {
 }
 
+message ClearConversationsMsgReq {
+  repeated string conversationIDs = 1;
+  string userID = 2;
+}
+
+message ClearConversationsMsgResp {
+}
+
+message UserClearAllMsgReq {
+  string userID = 1;
+}
+
+message UserClearAllMsgResp {
+}
+
+message DeleteMsgsReq {
+  string conversationID = 1;
+  repeated int64 seqs = 2;
+  string userID = 3;
+}
+
+message DeleteMsgsResp {
+}
+
+message DeleteMsgPhysicalReq {
+  repeated string conversationIDs = 1;
+  int64 remainTime = 2;
+}
+
+message DeleteMsgPhysicalResp {
+}
+
+message DeleteMsgPhysicalBySeqReq {
+  string conversationID = 1;
+  repeated int64 seqs = 2;
+}
+
+message DeleteMsgPhysicalBySeqResp {
+}
+
 service msg {
   //获取最小最大seq（包括用户的，以及指定群组的）
   rpc GetMaxSeq(sdkws.GetMaxSeqReq) returns(sdkws.GetMaxSeqResp);
@@ -212,12 +215,18 @@ service msg {
   rpc PullMessageBySeqs(sdkws.PullMessageBySeqsReq) returns(sdkws.PullMessageBySeqsResp);
   //发送消息
   rpc SendMsg(SendMsgReq) returns(SendMsgResp);
-  //删除某人消息
+
-  rpc DelMsgs(DelMsgsReq) returns(DelMsgsResp);
+  // 全量清空指定会话消息 重置min seq 比最大seq大1
-  //删除某个用户某个大群消息
+  rpc ClearConversationsMsg(ClearConversationsMsgReq) returns(ClearConversationsMsgResp);
-  rpc DelSuperGroupMsg(DelSuperGroupMsgReq) returns(DelSuperGroupMsgResp);
+  // 删除用户全部消息 重置min seq 比最大seq大1
-  //清空某人所有消息
+  rpc UserClearAllMsg(UserClearAllMsgReq) returns(UserClearAllMsgResp);
-  rpc ClearMsg(ClearMsgReq) returns(ClearMsgResp);
+  // 用户标记删除部分消息by Seq
+  rpc DeleteMsgs(DeleteMsgsReq) returns(DeleteMsgsResp);
+  // seq物理删除消息
+  rpc DeleteMsgPhysicalBySeq(DeleteMsgPhysicalBySeqReq) returns(DeleteMsgPhysicalBySeqResp);
+  // 物理删除消息by 时间
+  rpc DeleteMsgPhysical(DeleteMsgPhysicalReq) returns(DeleteMsgPhysicalResp);
+
   //设置消息是否发送成功-针对api发送的消息
   rpc SetSendMsgStatus(SetSendMsgStatusReq) returns(SetSendMsgStatusResp);
   //获取消息发送状态