mark as read

pkg/common/db/cache/msg.go

@@ -61,7 +61,10 @@ type SeqCache interface {
 	SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) error
 	// has read seq
 	SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
+	// k: user, v: seq
+	SetHasReadSeqs(ctx context.Context, conversationID string, hasReadSeqs map[string]int64) error
 	GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error)
+	GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error)
 }
 
 type thirdCache interface {
@@ -236,12 +239,22 @@ func (c *msgCache) SetHasReadSeq(ctx context.Context, userID string, conversatio
 	return utils.Wrap1(c.rdb.Set(ctx, c.getHasReadSeqKey(conversationID, userID), hasReadSeq, 0).Err())
 }
 
+func (c *msgCache) SetHasReadSeqs(ctx context.Context, conversationID string, hasReadSeqs map[string]int64) error {
+	return c.setSeqs(ctx, hasReadSeqs, func(userID string) string {
+		return c.getHasReadSeqKey(conversationID, userID)
+	})
+}
+
 func (c *msgCache) GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error) {
 	return c.getSeqs(ctx, conversationIDs, func(conversationID string) string {
 		return c.getHasReadSeqKey(conversationID, userID)
 	})
 }
 
+func (c *msgCache) GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error) {
+	return utils.Wrap2(c.rdb.Get(ctx, c.getHasReadSeqKey(conversationID, userID)).Int64())
+}
+
 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

@@ -343,9 +343,11 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
 		isNew = true
 	}
 	lastMaxSeq := currentMaxSeq
+	userSeqMap := make(map[string]int64)
 	for _, m := range msgs {
 		currentMaxSeq++
 		m.Seq = currentMaxSeq
+		userSeqMap[m.SendID] = m.Seq
 	}
 	failedNum, err := db.cache.SetMessageToCache(ctx, conversationID, msgs)
 	if err != nil {
@@ -360,6 +362,13 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
 	} else {
 		prome.Inc(prome.SeqSetSuccessCounter)
 	}
+	err2 := db.cache.SetHasReadSeqs(ctx, conversationID, userSeqMap)
+	if err != nil {
+		log.ZError(ctx, "SetHasReadSeqs error", err2, "userSeqMap", userSeqMap, "conversationID", conversationID)
+		prome.Inc(prome.SeqSetFailedCounter)
+	} else {
+		prome.Inc(prome.SeqSetSuccessCounter)
+	}
 	return lastMaxSeq, isNew, utils.Wrap(err, "")
 }