Merge branch 'ver3' of github.com:OpenIMSDK/Open-IM-Server into ver3

internal/msggateway/client.go

@@ -212,6 +212,7 @@ func (c *Client) PushMessage(ctx context.Context, msgData *sdkws.MsgData) error
 	} else {
 		msg.Msgs = m
 	}
+	log.ZDebug(ctx, "PushMessage", "msg", msg)
 	data, err := proto.Marshal(&msg)
 	if err != nil {
 		return err

internal/rpc/conversation/conversaion.go

@@ -210,6 +210,9 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbConver
 	if req.Conversation.IsPinned != nil {
 		m["is_pinned"] = req.Conversation.IsPinned.Value
 	}
+	if req.Conversation.GroupAtType != nil {
+		m["group_at_type"] = req.Conversation.GroupAtType.Value
+	}
 	if req.Conversation.IsPrivateChat != nil {
 		var conversations []*tableRelation.ConversationModel
 		for _, ownerUserID := range req.UserIDs {

internal/rpc/msg/revoke.go

@@ -3,6 +3,7 @@ package msg
 import (
 	"context"
 	"encoding/json"
+	"github.com/google/uuid"
 	"time"
 
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
@@ -40,15 +41,19 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	if len(msgs) == 0 || msgs[0] == nil {
 		return nil, errs.ErrRecordNotFound.Wrap("msg not found")
 	}
-	// todo: 判断是否已经撤回
+	if msgs[0].ContentType == constant.MsgRevokeNotification {
+		return nil, errs.ErrMsgAlreadyRevoke.Wrap("msg already revoke")
+	}
 	data, _ := json.Marshal(msgs[0])
 	log.ZInfo(ctx, "GetMsgBySeqs", "conversationID", req.ConversationID, "seq", req.Seq, "msg", string(data))
+	var role int32
 	if !tokenverify.IsAppManagerUid(ctx) {
 		switch msgs[0].SessionType {
 		case constant.SingleChatType:
 			if err := tokenverify.CheckAccessV3(ctx, msgs[0].SendID); err != nil {
 				return nil, err
 			}
+			role = user.AppMangerLevel
 		case constant.SuperGroupChatType:
 			members, err := m.Group.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID}), true)
 			if err != nil {
@@ -65,23 +70,28 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 					return nil, errs.ErrNoPermission.Wrap("no permission")
 				}
 			}
+			if member := members[req.UserID]; member != nil {
+				role = member.RoleLevel
+			}
 		default:
 			return nil, errs.ErrInternalServer.Wrap("msg sessionType not supported")
 		}
 	}
+	now := time.Now().UnixMilli()
 	err = m.MsgDatabase.RevokeMsg(ctx, req.ConversationID, req.Seq, &unRelationTb.RevokeModel{
+		ID:       uuid.New().String(),
+		Role:     role,
 		UserID:   req.UserID,
 		Nickname: user.Nickname,
-		Time:     time.Now().UnixMilli(),
+		Time:     now,
 	})
 	if err != nil {
 		return nil, err
 	}
-
 	tips := sdkws.RevokeMsgTips{
 		RevokerUserID:  req.UserID,
-		ClientMsgID:    "",
+		ClientMsgID:    msgs[0].ClientMsgID,
-		RevokeTime:     utils.GetCurrentTimestampByMill(),
+		RevokeTime:     now,
 		Seq:            req.Seq,
 		SesstionType:   msgs[0].SessionType,
 		ConversationID: req.ConversationID,

internal/rpc/msg/send.go

@@ -2,12 +2,15 @@ package msg
 
 import (
 	"context"
-
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
 	promePkg "github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
+	pbConversation "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/conversation"
 	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/proto/wrapperspb"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
 )
 
@@ -46,6 +49,9 @@ func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *pbMsg.SendMs
 	if err != nil {
 		return nil, err
 	}
+	if req.MsgData.ContentType == constant.AtText {
+		go m.setConversationAtInfo(ctx, req.MsgData)
+	}
 	if err = callbackAfterSendGroupMsg(ctx, req); err != nil {
 		log.ZWarn(ctx, "CallbackAfterSendGroupMsg", err)
 	}
@@ -56,6 +62,47 @@ func (m *msgServer) sendMsgSuperGroupChat(ctx context.Context, req *pbMsg.SendMs
 	resp.ClientMsgID = req.MsgData.ClientMsgID
 	return resp, nil
 }
+func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgData) {
+	log.ZDebug(nctx, "setConversationAtInfo", "msg", msg)
+	ctx := mcontext.NewCtx("@@@" + mcontext.GetOperationID(nctx))
+	var atUserID []string
+	conversation := &pbConversation.ConversationReq{
+		ConversationID:   utils.GetConversationIDByMsg(msg),
+		ConversationType: msg.SessionType,
+		GroupID:          msg.GroupID,
+	}
+	tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList)
+	if tagAll {
+		memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
+		if err != nil {
+			log.ZWarn(ctx, "GetGroupMemberIDs", err)
+			return
+		}
+		atUserID = utils.DifferenceString([]string{constant.AtAllString}, msg.AtUserIDList)
+		if len(atUserID) == 0 { //just @everyone
+			conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
+		} else { //@Everyone and @other people
+			conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAllAtMe}
+			err := m.Conversation.SetConversations(ctx, atUserID, conversation)
+			if err != nil {
+				log.ZWarn(ctx, "SetConversations", err, "userID", atUserID, "conversation", conversation)
+			}
+			memberUserIDList = utils.DifferenceString(atUserID, memberUserIDList)
+		}
+		conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtAll}
+		err = m.Conversation.SetConversations(ctx, memberUserIDList, conversation)
+		if err != nil {
+			log.ZWarn(ctx, "SetConversations", err, "userID", memberUserIDList, "conversation", conversation)
+		}
+	} else {
+		conversation.GroupAtType = &wrapperspb.Int32Value{Value: constant.AtMe}
+		err := m.Conversation.SetConversations(ctx, msg.AtUserIDList, conversation)
+		if err != nil {
+			log.ZWarn(ctx, "SetConversations", err, msg.AtUserIDList, conversation)
+		}
+	}
+
+}
 
 func (m *msgServer) sendMsgNotification(ctx context.Context, req *pbMsg.SendMsgReq) (resp *pbMsg.SendMsgResp, err error) {
 	promePkg.Inc(promePkg.SingleChatMsgRecvSuccessCounter)

pkg/common/db/table/unrelation/msg.go

@@ -21,6 +21,8 @@ type MsgDocModel struct {
 }
 
 type RevokeModel struct {
+	ID       string `bson:"id"`
+	Role     int32  `bson:"role"`
 	UserID   string `bson:"user_id"`
 	Nickname string `bson:"nickname"`
 	Time     int64  `bson:"time"`

pkg/common/db/unrelation/msg.go

@@ -246,24 +246,30 @@ func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, userID stri
 			continue
 		}
 		if msg.Revoke != nil {
-			// var conversationID string
-			// if index := strings.LastIndex(docID, ":"); index > 0 {
-			// 	conversationID = docID[:index]
-			// }
 			revokeContent := sdkws.MessageRevokedContent{
-				// RevokerID:      msg.Revoke.UserID,
+				RevokerID:                   msg.Revoke.ID,
-				// ClientMsgID:    msg.Msg.ClientMsgID,
+				RevokerRole:                 msg.Revoke.Role,
-				// RevokeTime:     msg.Revoke.Time,
+				ClientMsgID:                 msg.Msg.ClientMsgID,
-				// SesstionType:   msg.Msg.SessionType,
+				RevokerNickname:             msg.Revoke.Nickname,
-				// Seq:            msg.Msg.Seq,
+				RevokeTime:                  msg.Revoke.Time,
-				// ConversationID: conversationID,
+				SourceMessageSendTime:       msg.Msg.SendTime,
-				// ex:             msg.Msg.Ex,
+				SourceMessageSendID:         msg.Msg.SendID,
-			}
+				SourceMessageSenderNickname: msg.Msg.SenderNickname,
-			data, _ := json.Marshal(&revokeContent)
+				SessionType:                 msg.Msg.SessionType,
+				Seq:                         msg.Msg.Seq,
+				Ex:                          msg.Msg.Ex,
+			}
+			data, err := json.Marshal(&revokeContent)
+			if err != nil {
+				return nil, err
+			}
 			elem := sdkws.NotificationElem{
 				Detail: string(data),
 			}
-			content, _ := json.Marshal(&elem)
+			content, err := json.Marshal(&elem)
+			if err != nil {
+				return nil, err
+			}
 			msg.Msg.ContentType = constant.MsgRevokeNotification
 			msg.Msg.Content = string(content)
 		}

pkg/errs/code.go

@@ -68,6 +68,7 @@ const (
 	MutedInGroup          = 1402 //群成员被禁言
 	MutedGroup            = 1403 //群被禁言
 	UserNotRecvMsg        = 1404 //用户设置了不接收消息
+	MsgAlreadyRevoke      = 1405 //消息已撤回
 
 	// token错误码
 	TokenExpiredError             = 1501

pkg/errs/predefine.go

@@ -52,6 +52,7 @@ var (
 	ErrMutedInGroup     = NewCodeError(MutedInGroup, "MutedInGroup")
 	ErrMutedGroup       = NewCodeError(MutedGroup, "MutedGroup")
 	ErrUserNotRecvMsg   = NewCodeError(UserNotRecvMsg, "UserNotRecvMsg")
+	ErrMsgAlreadyRevoke = NewCodeError(MsgAlreadyRevoke, "MsgAlreadyRevoke")
 
 	ErrConnOverMaxNumLimit = NewCodeError(ConnOverMaxNumLimit, "ConnOverMaxNumLimit")
 

pkg/rpcclient/conversation.go

@@ -63,6 +63,10 @@ func (c *ConversationRpcClient) SetConversationMaxSeq(ctx context.Context, owner
 	_, err := c.Client.SetConversationMaxSeq(ctx, &pbConversation.SetConversationMaxSeqReq{OwnerUserID: ownerUserIDs, ConversationID: conversationID, MaxSeq: maxSeq})
 	return err
 }
+func (c *ConversationRpcClient) SetConversations(ctx context.Context, userIDs []string, conversation *pbConversation.ConversationReq) error {
+	_, err := c.Client.SetConversations(ctx, &pbConversation.SetConversationsReq{UserIDs: userIDs, Conversation: conversation})
+	return err
+}
 
 func (c *ConversationRpcClient) GetConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
 	resp, err := c.Client.GetConversationIDs(ctx, &pbConversation.GetConversationIDsReq{UserID: ownerUserID})