mark as read and modify pb

pkg/common/db/controller/msg.go

@@ -68,6 +68,7 @@ type CommonMsgDatabase interface {
 	SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) (err error)
 	SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
 	GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error)
+	GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, 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)
@@ -474,7 +475,6 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
 			maxSeq = userMaxSeq
 		}
 	}
-
 	if begin < minSeq {
 		begin = minSeq
 	}
@@ -810,6 +810,10 @@ func (db *commonMsgDatabase) GetHasReadSeqs(ctx context.Context, userID string,
 	return db.cache.GetHasReadSeqs(ctx, userID, conversationIDs)
 }
 
+func (db *commonMsgDatabase) GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error) {
+	return db.cache.GetHasReadSeq(ctx, userID, conversationID)
+}
+
 func (db *commonMsgDatabase) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
 	return db.cache.SetSendMsgStatus(ctx, id, status)
 }

pkg/common/db/unrelation/msg.go

@@ -281,16 +281,25 @@ func (m *MsgMongoDriver) IsExistDocID(ctx context.Context, docID string) (bool,
 }
 
 func (m *MsgMongoDriver) MarkSingleChatMsgsAsRead(ctx context.Context, userID string, docID string, indexes []int64) error {
-	updates := bson.M{
-		"$set": bson.M{},
-	}
+	updates := []mongo.WriteModel{}
 	for _, index := range indexes {
-		updates["$set"].(bson.M)[fmt.Sprintf("msgs.%d.is_read", index)] = true
+		filter := bson.M{
+			"doc_id": docID,
+			fmt.Sprintf("msgs.%d.msg.send_id", index): bson.M{
+				"$ne": userID,
+			},
+		}
+		update := bson.M{
+			"$set": bson.M{
+				fmt.Sprintf("msgs.%d.msg.is_read", index): true,
+			},
+		}
+		updateModel := mongo.NewUpdateManyModel().
+			SetFilter(filter).
+			SetUpdate(update)
+		updates = append(updates, updateModel)
+
 	}
-	filter := bson.M{"doc_id": docID}
-	_, err := m.MsgCollection.UpdateMany(ctx, filter, updates)
-	if err != nil {
-		return utils.Wrap(err, "")
-	}
-	return nil
+	_, err := m.MsgCollection.BulkWrite(ctx, updates)
+	return err
 }