Merge branch 'openimsdk:main' into main

.golangci.yml

@@ -171,7 +171,7 @@ linters-settings:
   #   exclude_godoc_examples: false
   funlen:
     lines: 150
-    statements: 50
+    statements: 80
   gci:
     # put imports beginning with prefix after 3rd-party packages;
     # only support one prefix

cmd/openim-api/main.go

@@ -26,13 +26,13 @@ import (
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"github.com/OpenIMSDK/tools/log"
+
 	"github.com/openimsdk/open-im-server/v3/internal/api"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 )
 
-
 func main() {
 	apiCmd := cmd.NewApiCmd()
 	apiCmd.AddPortFlag()

internal/api/msg.go

@@ -57,6 +57,10 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
 	var newContent string
 	options := make(map[string]bool, 5)
 	switch params.ContentType {
+	case constant.OANotification:
+		notification := sdkws.NotificationElem{}
+		notification.Detail = utils.StructToJsonString(params.Content)
+		newContent = utils.StructToJsonString(&notification)
 	case constant.Text:
 		fallthrough
 	case constant.Picture:
@@ -69,10 +73,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
 		fallthrough
 	case constant.File:
 		fallthrough
-	case constant.CustomNotTriggerConversation:
-		fallthrough
-	case constant.CustomOnlineOnly:
-		fallthrough
 	default:
 		newContent = utils.StructToJsonString(params.Content)
 	}
@@ -82,11 +82,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
 	if params.NotOfflinePush {
 		utils.SetSwitchFromOptions(options, constant.IsOfflinePush, false)
 	}
-	if params.ContentType == constant.CustomOnlineOnly {
-		m.SetOptions(options, false)
-	} else if params.ContentType == constant.CustomNotTriggerConversation {
-		utils.SetSwitchFromOptions(options, constant.IsConversationUpdate, false)
-	}
 	pbData := msg.SendMsgReq{
 		MsgData: &sdkws.MsgData{
 			SendID:           params.SendID,
@@ -105,14 +100,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
 			OfflinePushInfo:  params.OfflinePushInfo,
 		},
 	}
-	//if params.ContentType == constant.OANotification {
-	//	var tips sdkws.TipsComm
-	//	tips.JsonDetail = utils.StructToJsonString(params.Content)
-	//	pbData.MsgData.Content, err = proto.Marshal(&tips)
-	//	if err != nil {
-	//		log.ZError(c, "Marshal failed ", err, "tips", tips.String())
-	//	}
-	//}
 	return &pbData
 }
 
@@ -180,15 +167,13 @@ func (m *MessageApi) getSendMsgReq(c *gin.Context, req apistruct.SendMsg) (sendM
 		data = apistruct.FileElem{}
 	case constant.Custom:
 		data = apistruct.CustomElem{}
-	case constant.Revoke:
-		data = apistruct.RevokeElem{}
 	case constant.OANotification:
 		data = apistruct.OANotificationElem{}
 		req.SessionType = constant.NotificationChatType
-	case constant.CustomNotTriggerConversation:
-		data = apistruct.CustomElem{}
-	case constant.CustomOnlineOnly:
-		data = apistruct.CustomElem{}
+		if !authverify.IsManagerUserID(req.SendID) {
+			return nil, errs.ErrNoPermission.
+				Wrap("only app manager can as sender send OANotificationElem")
+		}
 	default:
 		return nil, errs.ErrArgs.WithDetail("not support err contentType")
 	}
@@ -212,7 +197,6 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
 		apiresp.GinError(c, errs.ErrNoPermission.Wrap("only app manager can send message"))
 		return
 	}
-
 	sendMsgReq, err := m.getSendMsgReq(c, req.SendMsg)
 	if err != nil {
 		log.ZError(c, "decodeData failed", err)

internal/api/route.go

@@ -83,7 +83,6 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
 		userRouterGroup.POST("/get_users_online_status", ParseToken, u.GetUsersOnlineStatus)
 		userRouterGroup.POST("/get_users_online_token_detail", ParseToken, u.GetUsersOnlineTokenDetail)
 		userRouterGroup.POST("/subscribe_users_status", ParseToken, u.SubscriberStatus)
-		userRouterGroup.POST("/unsubscribe_users_status", ParseToken, u.UnSubscriberStatus)
 		userRouterGroup.POST("/get_users_status", ParseToken, u.GetUserStatus)
 		userRouterGroup.POST("/get_subscribe_users_status", ParseToken, u.GetSubscribeUsersStatus)
 	}

internal/api/user.go

@@ -190,11 +190,6 @@ func (u *UserApi) SubscriberStatus(c *gin.Context) {
 	a2r.Call(user.UserClient.SubscribeOrCancelUsersStatus, u.Client, c)
 }
 
-// UnSubscriberStatus Unsubscribe a user's presence.
-func (u *UserApi) UnSubscriberStatus(c *gin.Context) {
-	a2r.Call(user.UserClient.SubscribeOrCancelUsersStatus, u.Client, c)
-}
-
 // GetUserStatus Get the online status of the user.
 func (u *UserApi) GetUserStatus(c *gin.Context) {
 	a2r.Call(user.UserClient.GetUserStatus, u.Client, c)

internal/msgtransfer/init.go

@@ -16,12 +16,13 @@ package msgtransfer
 
 import (
 	"fmt"
-
 	"sync"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
+
 	"github.com/OpenIMSDK/tools/mw"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -30,7 +31,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prome"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )

internal/msgtransfer/online_history_msg_handler.go

@@ -283,20 +283,30 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(
 			return
 		}
 		if isNewConversation {
-			if storageList[0].SessionType == constant.SuperGroupChatType {
-				log.ZInfo(ctx, "group chat first create conversation", "conversationID", conversationID)
+			switch storageList[0].SessionType {
+			case constant.SuperGroupChatType:
+				log.ZInfo(ctx, "group chat first create conversation", "conversationID",
+					conversationID)
 				userIDs, err := och.groupRpcClient.GetGroupMemberIDs(ctx, storageList[0].GroupID)
 				if err != nil {
-					log.ZWarn(ctx, "get group member ids error", err, "conversationID", conversationID)
+					log.ZWarn(ctx, "get group member ids error", err, "conversationID",
+						conversationID)
 				} else {
-					if err := och.conversationRpcClient.GroupChatFirstCreateConversation(ctx, storageList[0].GroupID, userIDs); err != nil {
-						log.ZWarn(ctx, "single chat first create conversation error", err, "conversationID", conversationID)
+					if err := och.conversationRpcClient.GroupChatFirstCreateConversation(ctx,
+						storageList[0].GroupID, userIDs); err != nil {
+						log.ZWarn(ctx, "single chat first create conversation error", err,
+							"conversationID", conversationID)
 					}
 				}
-			} else {
-				if err := och.conversationRpcClient.SingleChatFirstCreateConversation(ctx, storageList[0].RecvID, storageList[0].SendID); err != nil {
-					log.ZWarn(ctx, "single chat first create conversation error", err, "conversationID", conversationID)
+			case constant.SingleChatType, constant.NotificationChatType:
+				if err := och.conversationRpcClient.SingleChatFirstCreateConversation(ctx, storageList[0].RecvID,
+					storageList[0].SendID, conversationID, storageList[0].SessionType); err != nil {
+					log.ZWarn(ctx, "single chat or notification first create conversation error", err,
+						"conversationID", conversationID, "sessionType", storageList[0].SessionType)
 				}
+			default:
+				log.ZWarn(ctx, "unknown session type", nil, "sessionType",
+					storageList[0].SessionType)
 			}
 		}
 

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -62,7 +62,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(
 		log.ZError(ctx, "msgFromMQ.MsgData is empty", nil, "cMsg", cMsg)
 		return
 	}
-	log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.MsgData)
+	log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String())
 	err = mc.msgDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
 	if err != nil {
 		log.ZError(

internal/push/push_handler.go

@@ -58,6 +58,7 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 	}
 	sec := msgFromMQ.MsgData.SendTime / 1000
 	nowSec := utils.GetCurrentTimestampBySecond()
+	log.ZDebug(ctx, "push msg", "msg", pbData.String(), "sec", sec, "nowSec", nowSec)
 	if nowSec-sec > 10 {
 		return
 	}

internal/push/push_to_client.go

@@ -126,13 +126,12 @@ func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.Msg
 }
 
 func (p *Pusher) UnmarshalNotificationElem(bytes []byte, t interface{}) error {
-	var notificationElem struct {
-		Detail string `json:"detail,omitempty"`
-	}
-	if err := json.Unmarshal(bytes, &notificationElem); err != nil {
+	var notification sdkws.NotificationElem
+	if err := json.Unmarshal(bytes, &notification); err != nil {
 		return err
 	}
-	return json.Unmarshal([]byte(notificationElem.Detail), t)
+
+	return json.Unmarshal([]byte(notification.Detail), t)
 }
 
 func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) {

internal/rpc/conversation/conversaion.go

@@ -17,8 +17,6 @@ package conversation
 import (
 	"context"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
-
 	"google.golang.org/grpc"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -114,7 +112,10 @@ func (c *conversationServer) SetConversation(ctx context.Context, req *pbconvers
 	return resp, nil
 }
 
-func (c *conversationServer) SetConversations(ctx context.Context, req *pbconversation.SetConversationsReq) (*pbconversation.SetConversationsResp, error) {
+//nolint
+func (c *conversationServer) SetConversations(ctx context.Context,
+	req *pbconversation.SetConversationsReq,
+) (*pbconversation.SetConversationsResp, error) {
 	if req.Conversation == nil {
 		return nil, errs.ErrArgs.Wrap("conversation must not be nil")
 	}
@@ -124,14 +125,8 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
 			return nil, err
 		}
 		if groupInfo.Status == constant.GroupStatusDismissed {
-			return nil, err
+			return nil, errs.ErrDismissedAlready.Wrap("group dismissed")
 		}
-		// for _, userID := range req.UserIDs {
-		// 	if _, err := c.groupRpcClient.GetGroupMemberCache(ctx, req.Conversation.GroupID, userID); err != nil {
-		// 		log.ZError(ctx, "user not in group", err, "userID", userID, "groupID", req.Conversation.GroupID)
-		// 		return nil, err
-		// 	}
-		// }
 	}
 	var unequal int
 	var conv tablerelation.ConversationModel
@@ -205,7 +200,14 @@ func (c *conversationServer) SetConversations(ctx context.Context, req *pbconver
 			return nil, err
 		}
 		for _, userID := range req.UserIDs {
-			c.conversationNotificationSender.ConversationSetPrivateNotification(ctx, userID, req.Conversation.UserID, req.Conversation.IsPrivateChat.Value, req.Conversation.ConversationID)
+			err := c.conversationNotificationSender.ConversationSetPrivateNotification(ctx, userID, req.Conversation.UserID,
+				req.Conversation.IsPrivateChat.Value, req.Conversation.ConversationID)
+			if err != nil {
+				log.ZWarn(ctx, "send conversation set private notification failed", err,
+					"userID", userID, "conversationID", req.Conversation.ConversationID)
+
+				continue
+			}
 		}
 	}
 	if req.Conversation.BurnDuration != nil {
@@ -235,24 +237,40 @@ func (c *conversationServer) GetRecvMsgNotNotifyUserIDs(ctx context.Context, req
 }
 
 // create conversation without notification for msg redis transfer.
-func (c *conversationServer) CreateSingleChatConversations(ctx context.Context, req *pbconversation.CreateSingleChatConversationsReq) (*pbconversation.CreateSingleChatConversationsResp, error) {
-	var conversation tablerelation.ConversationModel
-	conversation.ConversationID = msgprocessor.GetConversationIDBySessionType(constant.SingleChatType, req.RecvID, req.SendID)
-	conversation.ConversationType = constant.SingleChatType
-	conversation.OwnerUserID = req.SendID
-	conversation.UserID = req.RecvID
-	err := c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation})
-	if err != nil {
-		log.ZWarn(ctx, "create conversation failed", err, "conversation", conversation)
-	}
+func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
+	req *pbconversation.CreateSingleChatConversationsReq,
+) (*pbconversation.CreateSingleChatConversationsResp, error) {
+	switch req.ConversationType {
+	case constant.SingleChatType:
+		var conversation tablerelation.ConversationModel
+		conversation.ConversationID = req.ConversationID
+		conversation.ConversationType = req.ConversationType
+		conversation.OwnerUserID = req.SendID
+		conversation.UserID = req.RecvID
+		err := c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation})
+		if err != nil {
+			log.ZWarn(ctx, "create conversation failed", err, "conversation", conversation)
+		}
 
-	conversation2 := conversation
-	conversation2.OwnerUserID = req.RecvID
-	conversation2.UserID = req.SendID
-	err = c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation2})
-	if err != nil {
-		log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation)
+		conversation2 := conversation
+		conversation2.OwnerUserID = req.RecvID
+		conversation2.UserID = req.SendID
+		err = c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation2})
+		if err != nil {
+			log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation)
+		}
+	case constant.NotificationChatType:
+		var conversation tablerelation.ConversationModel
+		conversation.ConversationID = req.ConversationID
+		conversation.ConversationType = req.ConversationType
+		conversation.OwnerUserID = req.RecvID
+		conversation.UserID = req.SendID
+		err := c.conversationDatabase.CreateConversation(ctx, []*tablerelation.ConversationModel{&conversation})
+		if err != nil {
+			log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation)
+		}
 	}
+
 	return &pbconversation.CreateSingleChatConversationsResp{}, nil
 }
 

internal/rpc/friend/friend.go

@@ -252,7 +252,8 @@ func (s *friendServer) GetDesignatedFriends(
 	return resp, nil
 }
 
-func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context, req *pbfriend.GetDesignatedFriendsApplyReq) (resp *pbfriend.GetDesignatedFriendsApplyResp, err error) {
+func (s *friendServer) GetDesignatedFriendsApply(ctx context.Context,
+	req *pbfriend.GetDesignatedFriendsApplyReq) (resp *pbfriend.GetDesignatedFriendsApplyResp, err error) {
 	friendRequests, err := s.friendDatabase.FindBothFriendRequests(ctx, req.FromUserID, req.ToUserID)
 	if err != nil {
 		return nil, err

internal/rpc/group/group.go

@@ -154,16 +154,16 @@ func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error
 	return nil
 }
 
-func (s *groupServer) GetUsernameMap(ctx context.Context, userIDs []string, complete bool) (map[string]string, error) {
+func (s *groupServer) GetPublicUserInfoMap(ctx context.Context, userIDs []string, complete bool) (map[string]*sdkws.PublicUserInfo, error) {
 	if len(userIDs) == 0 {
-		return map[string]string{}, nil
+		return map[string]*sdkws.PublicUserInfo{}, nil
 	}
 	users, err := s.User.GetPublicUserInfos(ctx, userIDs, complete)
 	if err != nil {
 		return nil, err
 	}
-	return utils.SliceToMapAny(users, func(e *sdkws.PublicUserInfo) (string, string) {
-		return e.UserID, e.Nickname
+	return utils.SliceToMapAny(users, func(e *sdkws.PublicUserInfo) (string, *sdkws.PublicUserInfo) {
+		return e.UserID, e
 	}), nil
 }
 
@@ -468,15 +468,18 @@ func (s *groupServer) GetGroupAllMember(ctx context.Context, req *pbgroup.GetGro
 	if err != nil {
 		return nil, err
 	}
-	nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
-		return e.UserID, e.Nickname == ""
+	publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
+		return e.UserID, e.Nickname == "" || e.FaceURL == ""
 	}), true)
 	if err != nil {
 		return nil, err
 	}
 	resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo {
 		if e.Nickname == "" {
-			e.Nickname = nameMap[e.UserID]
+			e.Nickname = publicUserInfoMap[e.UserID].Nickname
+		}
+		if e.FaceURL == "" {
+			e.FaceURL = publicUserInfoMap[e.UserID].FaceURL
 		}
 		return convert.Db2PbGroupMember(e)
 	})
@@ -616,15 +619,18 @@ func (s *groupServer) GetGroupMembersInfo(ctx context.Context, req *pbgroup.GetG
 	if err != nil {
 		return nil, err
 	}
-	nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
-		return e.UserID, e.Nickname == ""
+	publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
+		return e.UserID, e.Nickname == "" || e.FaceURL == ""
 	}), true)
 	if err != nil {
 		return nil, err
 	}
 	resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo {
 		if e.Nickname == "" {
-			e.Nickname = nameMap[e.UserID]
+			e.Nickname = publicUserInfoMap[e.UserID].Nickname
+		}
+		if e.FaceURL == "" {
+			e.FaceURL = publicUserInfoMap[e.UserID].FaceURL
 		}
 		return convert.Db2PbGroupMember(e)
 	})
@@ -1067,15 +1073,18 @@ func (s *groupServer) GetGroupMembersCMS(ctx context.Context, req *pbgroup.GetGr
 		return nil, err
 	}
 	resp.Total = total
-	nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
-		return e.UserID, e.Nickname == ""
+	nameMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
+		return e.UserID, e.Nickname == "" || e.FaceURL == ""
 	}), true)
 	if err != nil {
 		return nil, err
 	}
 	resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo {
 		if e.Nickname == "" {
-			e.Nickname = nameMap[e.UserID]
+			e.Nickname = nameMap[e.UserID].Nickname
+		}
+		if e.FaceURL == "" {
+			e.FaceURL = nameMap[e.UserID].FaceURL
 		}
 		return convert.Db2PbGroupMember(e)
 	})
@@ -1461,7 +1470,7 @@ func (s *groupServer) GetUserInGroupMembers(ctx context.Context, req *pbgroup.Ge
 	if err != nil {
 		return nil, err
 	}
-	nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
+	publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
 		return e.UserID, e.Nickname == ""
 	}), true)
 	if err != nil {
@@ -1469,7 +1478,10 @@ func (s *groupServer) GetUserInGroupMembers(ctx context.Context, req *pbgroup.Ge
 	}
 	resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo {
 		if e.Nickname == "" {
-			e.Nickname = nameMap[e.UserID]
+			e.Nickname = publicUserInfoMap[e.UserID].Nickname
+		}
+		if e.FaceURL == "" {
+			e.FaceURL = publicUserInfoMap[e.UserID].FaceURL
 		}
 		return convert.Db2PbGroupMember(e)
 	})
@@ -1494,15 +1506,18 @@ func (s *groupServer) GetGroupMemberRoleLevel(ctx context.Context, req *pbgroup.
 	if err != nil {
 		return nil, err
 	}
-	nameMap, err := s.GetUsernameMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
-		return e.UserID, e.Nickname == ""
+	publicUserInfoMap, err := s.GetPublicUserInfoMap(ctx, utils.Filter(members, func(e *relationtb.GroupMemberModel) (string, bool) {
+		return e.UserID, e.Nickname == "" || e.FaceURL == ""
 	}), true)
 	if err != nil {
 		return nil, err
 	}
 	resp.Members = utils.Slice(members, func(e *relationtb.GroupMemberModel) *sdkws.GroupMemberFullInfo {
 		if e.Nickname == "" {
-			e.Nickname = nameMap[e.UserID]
+			e.Nickname = publicUserInfoMap[e.UserID].Nickname
+		}
+		if e.FaceURL == "" {
+			e.FaceURL = publicUserInfoMap[e.UserID].FaceURL
 		}
 		return convert.Db2PbGroupMember(e)
 	})

internal/rpc/msg/as_read.go

@@ -147,7 +147,6 @@ func (m *msgServer) MarkConversationAsRead(
 		for i := hasReadSeq + 1; i <= req.HasReadSeq; i++ {
 			seqs = append(seqs, i)
 		}
-
 		if len(seqs) > 0 {
 			log.ZDebug(ctx, "MarkConversationAsRead", "seqs", seqs, "conversationID", req.ConversationID)
 			if err = m.MsgDatabase.MarkSingleChatMsgsAsRead(ctx, req.UserID, req.ConversationID, seqs); err != nil {
@@ -165,7 +164,9 @@ func (m *msgServer) MarkConversationAsRead(
 			m.conversationAndGetRecvID(conversation, req.UserID), seqs, hasReadSeq); err != nil {
 			return nil, err
 		}
-	} else if conversation.ConversationType == constant.SuperGroupChatType {
+
+	} else if conversation.ConversationType == constant.SuperGroupChatType ||
+		conversation.ConversationType == constant.NotificationChatType {
 		if req.HasReadSeq > hasReadSeq {
 			err = m.MsgDatabase.SetHasReadSeq(ctx, req.UserID, req.ConversationID, req.HasReadSeq)
 			if err != nil {

internal/tools/msg.go

@@ -23,6 +23,8 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
+
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/mcontext"
@@ -35,7 +37,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient/notification"
 )
@@ -77,7 +78,8 @@ func InitMsgTool() (*MsgTool, error) {
 	/*
 		discov, err := zookeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
 			zookeeper.WithFreq(time.Hour), zookeeper.WithRoundRobin(), zookeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,
-				config.Config.Zookeeper.Password), zookeeper.WithTimeout(10), zookeeper.WithLogger(log.NewZkLogger()))*/if err != nil {
+				config.Config.Zookeeper.Password), zookeeper.WithTimeout(10), zookeeper.WithLogger(log.NewZkLogger()))*/
+	if err != nil {
 		return nil, err
 	}
 	discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))

pkg/apistruct/manage.go

@@ -41,7 +41,7 @@ type SendMsgReq struct {
 type BatchSendMsgReq struct {
 	SendMsg
 	IsSendAll bool     `json:"isSendAll"`
-	RecvIDs   []string `json:"recvIDs"`
+	RecvIDs   []string `json:"recvIDs" binding:"required"`
 }
 
 type BatchSendMsgResp struct {

pkg/common/db/controller/msg.go

@@ -701,7 +701,14 @@ func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Cont
 	return db.cache.SetMinSeq(ctx, conversationID, minSeq)
 }
 
-func processMsgDocModel(ctx context.Context, msgDocModel *unrelationtb.MsgDocModel, userID, conversationID string, index int64, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, over bool) {
+func processMsgDocModel(
+	ctx context.Context,
+	msgDocModel *unrelationtb.MsgDocModel,
+	userID, conversationID string,
+	index int64,
+	destructTime int64,
+	lastMsgDestructTime time.Time,
+) (seqs []int64, over bool) {
 	if len(msgDocModel.Msg) > 0 {
 		i := 0
 		for _, msg := range msgDocModel.Msg {
@@ -823,7 +830,14 @@ func handleFullAndExpiredForDeleteMsgRecursion(ctx context.Context, msgDocModel
 	delStruct.minSeq = msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.Seq
 }
 
-func handleNotFullAndExpiredForDeleteMsgRecursion(ctx context.Context, msgDocModel *unrelationtb.MsgDocModel, remainTime, index int64, conversationID string, delStruct *delMsgRecursionStruct, db *commonMsgDatabase) {
+func handleNotFullAndExpiredForDeleteMsgRecursion(
+	ctx context.Context,
+	msgDocModel *unrelationtb.MsgDocModel,
+	remainTime, index int64,
+	conversationID string,
+	delStruct *delMsgRecursionStruct,
+	db *commonMsgDatabase,
+) {
 	var delMsgIndexs []int
 	for i, MsgInfoModel := range msgDocModel.Msg {
 		if MsgInfoModel != nil && MsgInfoModel.Msg != nil {

pkg/common/db/s3/minio/minio.go

@@ -430,7 +430,13 @@ func (m *Minio) presignedGetObject(ctx context.Context, name string, expire time
 	return rawURL.String(), nil
 }
 
-func (m *Minio) getImageInfoForAccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption, reqParams url.Values) (fileInfo *s3.ObjectInfo, objectInfoPath, msg string, err error) {
+func (m *Minio) getImageInfoForAccessURL(
+	ctx context.Context,
+	name string,
+	expire time.Duration,
+	opt *s3.AccessURLOption,
+	reqParams url.Values,
+) (fileInfo *s3.ObjectInfo, objectInfoPath, msg string, err error) {
 	if opt != nil {
 		if opt.ContentType != "" {
 			reqParams.Set("response-content-type", opt.ContentType)

pkg/common/db/table/relation/group.go

@@ -30,7 +30,7 @@ type GroupModel struct {
 	Introduction           string    `gorm:"column:introduction;size:255"                        json:"introduction"`
 	FaceURL                string    `gorm:"column:face_url;size:255"                            json:"faceURL"`
 	CreateTime             time.Time `gorm:"column:create_time;index:create_time;autoCreateTime"`
-	Ex                     string    `gorm:"column:ex;size:1024"                                           json:"ex"`
+	Ex                     string    `gorm:"column:ex;size:1024"                                 json:"ex"`
 	Status                 int32     `gorm:"column:status"`
 	CreatorUserID          string    `gorm:"column:creator_user_id;size:64"`
 	GroupType              int32     `gorm:"column:group_type"`

pkg/rpcclient/conversation.go

@@ -39,7 +39,6 @@ func NewConversation(discov discoveryregistry.SvcDiscoveryRegistry) *Conversatio
 		panic(err)
 	}
 	client := pbconversation.NewConversationClient(conn)
-
 	return &Conversation{discov: discov, conn: conn, Client: client}
 }
 
@@ -57,13 +56,17 @@ func (c *ConversationRpcClient) GetSingleConversationRecvMsgOpt(ctx context.Cont
 	if err != nil {
 		return 0, err
 	}
-
 	return conversation.GetConversation().RecvMsgOpt, err
 }
 
-func (c *ConversationRpcClient) SingleChatFirstCreateConversation(ctx context.Context, recvID, sendID string) error {
-	_, err := c.Client.CreateSingleChatConversations(ctx, &pbconversation.CreateSingleChatConversationsReq{RecvID: recvID, SendID: sendID})
-
+func (c *ConversationRpcClient) SingleChatFirstCreateConversation(ctx context.Context, recvID, sendID,
+	conversationID string, conversationType int32,
+) error {
+	_, err := c.Client.CreateSingleChatConversations(ctx,
+		&pbconversation.CreateSingleChatConversationsReq{
+			RecvID: recvID, SendID: sendID, ConversationID: conversationID,
+			ConversationType: conversationType,
+		})
 	return err
 }
 

pkg/rpcclient/msg.go

@@ -221,12 +221,13 @@ func WithRpcGetUserName() NotificationOptions {
 	}
 }
 
-func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, sendID, recvID string, contentType, sesstionType int32, m proto.Message, opts ...NotificationOptions) (err error) {
+func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, sendID, recvID string,
+	contentType, sesstionType int32, m proto.Message, opts ...NotificationOptions) (err error) {
 	n := sdkws.NotificationElem{Detail: utils.StructToJsonString(m)}
 	content, err := json.Marshal(&n)
 	if err != nil {
-		log.ZError(ctx, "MsgClient Notification json.Marshal failed", err, "sendID", sendID, "recvID", recvID, "contentType", contentType, "msg", m)
-
+		log.ZError(ctx, "MsgClient Notification json.Marshal failed", err, "sendID",
+			sendID, "recvID", recvID, "contentType", contentType, "msg", m)
 		return err
 	}
 	notificationOpt := &notificationOpt{}
@@ -235,8 +236,8 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s
 	}
 	var req msg.SendMsgReq
 	var msg sdkws.MsgData
+	var userInfo *sdkws.UserInfo
 	if notificationOpt.WithRpcGetUsername && s.getUserInfo != nil {
-		var userInfo *sdkws.UserInfo
 		userInfo, err = s.getUserInfo(ctx, sendID)
 		if err != nil {
 			log.ZWarn(ctx, "getUserInfo failed", err, "sendID", sendID)
@@ -259,7 +260,7 @@ func (s *NotificationSender) NotificationWithSesstionType(ctx context.Context, s
 	msg.CreateTime = utils.GetCurrentTimestampByMill()
 	msg.ClientMsgID = utils.GetMsgID(sendID)
 	optionsConfig := s.contentTypeConf[contentType]
-	if sesstionType == constant.SuperGroupChatType && contentType == constant.HasReadReceipt {
+	if sendID == recvID && contentType == constant.HasReadReceipt {
 		optionsConfig.ReliabilityLevel = constant.UnreliableNotification
 	}
 	options := config.GetOptionsByNotification(optionsConfig)