fix: optimize conversation message handling and improve message synchronization

Author: withchao

internal/conversation_msg/conversation_msg.go

@@ -8,6 +8,9 @@ import (
 	"math"
 	"sync"
 
+	"sort"
+	"time"
+
 	"github.com/openimsdk/openim-sdk-core/v3/internal/group"
 	"github.com/openimsdk/openim-sdk-core/v3/internal/interaction"
 	"github.com/openimsdk/openim-sdk-core/v3/internal/relation"
@@ -30,9 +33,6 @@ import (
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
 
-	"sort"
-	"time"
-
 	"github.com/openimsdk/openim-sdk-core/v3/pkg/utils"
 	"github.com/openimsdk/openim-sdk-core/v3/sdk_struct"
 )
@@ -225,6 +225,7 @@ func (c *Conversation) doMsgNew(c2v common.Cmd2Value) {
 
 	for conversationID, msgs := range allMsg {
 		conversationIDs = append(conversationIDs, conversationID)
+
 		log.ZDebug(ctx, "parse message in one conversation", "conversationID", conversationID, "message length", len(msgs.Msgs), "data", msgs.Msgs)
 
 		clientIDs := make([]string, 0, len(msgs.Msgs))
@@ -436,27 +437,30 @@ func (c *Conversation) doMsgNew(c2v common.Cmd2Value) {
 		}
 	}
 
-	if err := c.db.BatchUpdateConversationList(ctx, append(mapConversationToList(conversationChangedSet), mapConversationToList(phConversationChangedSet)...)); err != nil {
+	if err := c.db.BatchUpdateConversationList(ctx, append(datautil.Values(conversationChangedSet), datautil.Values(phConversationChangedSet)...)); err != nil {
 		log.ZError(ctx, "insert changed conversation err :", err)
 	}
 	//New conversation storage
 
-	if err := c.db.BatchInsertConversationList(ctx, mapConversationToList(phNewConversationSet)); err != nil {
+	if err := c.db.BatchInsertConversationList(ctx, datautil.Values(phNewConversationSet)); err != nil {
 		log.ZError(ctx, "insert new conversation err:", err)
 	}
 	log.ZDebug(ctx, "before trigger msg", "cost time", time.Since(b).Seconds(), "len", len(allMsg))
 
-	c.newMessage(ctx, newMessages, conversationChangedSet, newConversationSet, onlineMap)
-
+	if c.batchMsgListener() != nil {
+		c.batchNewMessages(ctx, newMessages, conversationChangedSet, newConversationSet, onlineMap)
+	} else {
+		c.newMessage(ctx, newMessages, conversationChangedSet, newConversationSet, onlineMap)
+	}
 	if len(newConversationSet) > 0 {
-		c.doUpdateConversation(common.Cmd2Value{Value: common.UpdateConNode{Action: constant.NewConDirect, Args: utils.StructToJsonString(mapConversationToList(newConversationSet))}})
+		c.ConversationListener().OnNewConversation(utils.StructToJsonString(datautil.Values(newConversationSet)))
 	}
 	if len(conversationChangedSet) > 0 {
-		c.doUpdateConversation(common.Cmd2Value{Value: common.UpdateConNode{Action: constant.ConChangeDirect, Args: utils.StructToJsonString(mapConversationToList(conversationChangedSet))}})
+		c.ConversationListener().OnConversationChanged(utils.StructToJsonString(datautil.Values(conversationChangedSet)))
 	}
 
 	if isTriggerUnReadCount {
-		c.doUpdateConversation(common.Cmd2Value{Value: common.UpdateConNode{Action: constant.TotalUnreadMessageChanged, Args: ""}})
+		_ = c.OnTotalUnreadMessageCountChanged(ctx)
 	}
 
 	for _, msgs := range allMsg {
@@ -474,6 +478,18 @@ func (c *Conversation) doMsgNew(c2v common.Cmd2Value) {
 	log.ZDebug(ctx, "insert msg", "duration", fmt.Sprintf("%dms", time.Since(b)), "len", len(allMsg))
 }
 
+func (c *Conversation) OnTotalUnreadMessageCountChanged(ctx context.Context) error {
+	log.ZInfo(ctx, "OnTotalUnreadMessageCountChanged", "caller", common.GetCaller(2))
+	totalUnreadCount, err := c.db.GetTotalUnreadMsgCountDB(ctx)
+	if err != nil {
+		log.ZWarn(ctx, "TotalUnreadMessageChanged GetTotalUnreadMsgCountDB err", err)
+	} else {
+		log.ZDebug(ctx, "TotalUnreadMessageChanged", "totalUnreadCount", totalUnreadCount)
+		c.ConversationListener().OnTotalUnreadMessageCountChanged(totalUnreadCount)
+	}
+	return nil
+}
+
 func (c *Conversation) doMsgSyncByReinstalled(c2v common.Cmd2Value) {
 	allMsg := c2v.Value.(sdk_struct.CmdMsgSyncInReinstall).Msgs
 	ctx := c2v.Ctx

internal/interaction/msg_sync.go

@@ -317,31 +317,53 @@ func (m *MsgSyncer) doPushMsg(ctx context.Context, push *sdkws.PushMessages) {
 	m.pushTriggerAndSync(ctx, push.NotificationMsgs, m.triggerNotification)
 }
 
-func (m *MsgSyncer) pushTriggerAndSync(ctx context.Context, pushMessages map[string]*sdkws.PullMsgs, triggerFunc func(ctx context.Context, msgs map[string]*sdkws.PullMsgs) error) {
+func (m *MsgSyncer) pushTriggerAndSync(ctx context.Context, pushMessages map[string]*sdkws.PullMsgs,
+	triggerFunc func(ctx context.Context, msgs map[string]*sdkws.PullMsgs) error) {
 	if len(pushMessages) == 0 {
 		return
 	}
+
 	needSyncSeqMap := make(map[string][2]int64)
-	var lastSeq int64
-	var storageMsgs []*sdkws.MsgData
+	res := make(map[string]*sdkws.PullMsgs)
+
 	for conversationID, msgs := range pushMessages {
+		var (
+			lastSeq     int64
+			storageMsgs []*sdkws.MsgData
+		)
+
 		for _, msg := range msgs.Msgs {
 			if msg.Seq == 0 {
-				_ = triggerFunc(ctx, map[string]*sdkws.PullMsgs{conversationID: {Msgs: []*sdkws.MsgData{msg}}})
+				_ = triggerFunc(ctx, map[string]*sdkws.PullMsgs{
+					conversationID: {Msgs: []*sdkws.MsgData{msg}},
+				})
 				continue
 			}
 			lastSeq = msg.Seq
 			storageMsgs = append(storageMsgs, msg)
 		}
-		if lastSeq == m.syncedMaxSeqs[conversationID]+int64(len(storageMsgs)) && lastSeq != 0 {
-			log.ZDebug(ctx, "trigger msgs", "msgs", storageMsgs)
-			_ = triggerFunc(ctx, map[string]*sdkws.PullMsgs{conversationID: {Msgs: storageMsgs}})
+
+		if len(storageMsgs) == 0 {
+			continue
+		}
+
+		expectedLast := m.syncedMaxSeqs[conversationID] + int64(len(storageMsgs))
+		if lastSeq == expectedLast {
+			log.ZDebug(ctx, "trigger msgs", "conversationID", conversationID, "msgs", storageMsgs)
+			res[conversationID] = &sdkws.PullMsgs{Msgs: storageMsgs}
 			m.syncedMaxSeqs[conversationID] = lastSeq
-		} else if lastSeq != 0 && lastSeq > m.syncedMaxSeqs[conversationID] {
-			//must pull message when message type is notification
-			needSyncSeqMap[conversationID] = [2]int64{m.syncedMaxSeqs[conversationID] + 1, lastSeq}
+		} else if lastSeq > m.syncedMaxSeqs[conversationID] {
+			// must pull message when message type is notification
+			needSyncSeqMap[conversationID] = [2]int64{
+				m.syncedMaxSeqs[conversationID] + 1,
+				lastSeq,
+			}
 		}
 	}
+
+	if len(res) > 0 {
+		_ = triggerFunc(ctx, res)
+	}
 	m.syncAndTriggerMsgs(ctx, needSyncSeqMap, defaultPullNums)
 }