feat: update progress logic. (#615)

* update trigger const.

* feat: implement more accurate progress.

* optimize method implement.

* update conversation List logic.

* update method logic.

* update conversation logic.

* update BatchInsertMessageList.

* update addProgress to private.

* fix update loss contents.

* update condition logic.

* fix: remove uncorrect use ctx.

* remove unnecesary content.

* fix:remove uncorrect log call.

* fix uncorrect progress add.

* recovery.

* fix: uncorrect map init.

* feat: update progress logic.

* fix: use map to correct.

* add debug log comment.

* update progress number implement.

* update SyncProgress.

Author: mo3et

internal/conversation_msg/conversation_msg.go

@@ -81,6 +81,7 @@ type Conversation struct {
 	full                 *full.Full
 	maxSeqRecorder       MaxSeqRecorder
 	IsExternalExtensions bool
+	msgOffset            int
 	progress             int
 
 	startTime time.Time
@@ -119,6 +120,7 @@ func NewConversation(ctx context.Context, longConnMgr *interaction.LongConnMgr,
 		messageController:    NewMessageController(db, ch),
 		IsExternalExtensions: info.IsExternalExtensions(),
 		maxSeqRecorder:       NewMaxSeqRecorder(),
+		msgOffset:            0,
 		progress:             0,
 	}
 	n.typing = newTyping(n)
@@ -448,12 +450,13 @@ func (c *Conversation) doMsgSyncByReinstalled(c2v common.Cmd2Value) {
 	allMsg := c2v.Value.(sdk_struct.CmdMsgSyncInReinstall).Msgs
 	ctx := c2v.Ctx
 	msgLen := len(allMsg)
+	c.msgOffset += msgLen
 	total := c2v.Value.(sdk_struct.CmdMsgSyncInReinstall).Total
 
 	insertMsg := make(map[string][]*model_struct.LocalChatLog, 10)
 	conversationList := make([]*model_struct.LocalConversation, 0)
 
-	log.ZDebug(ctx, "message come here conversation ch", "conversation length", msgLen)
+	log.ZDebug(ctx, "message come here conversation ch in reinstalled", "conversation length", msgLen)
 	b := time.Now()
 
 	for conversationID, msgs := range allMsg {
@@ -497,7 +500,7 @@ func (c *Conversation) doMsgSyncByReinstalled(c2v common.Cmd2Value) {
 			log.ZDebug(ctx, "decode message", "msg", msg)
 			if v.SendID == c.loginUserID {
 				// Messages sent by myself  //if  sent through  this terminal
-				log.ZInfo(ctx, "sync message", "msg", msg)
+				log.ZInfo(ctx, "sync message in reinstalled", "msg", msg)
 
 				latestMsg = msg
 
@@ -531,21 +534,17 @@ func (c *Conversation) doMsgSyncByReinstalled(c2v common.Cmd2Value) {
 	}
 	log.ZDebug(ctx, "before trigger msg", "cost time", time.Since(b).Seconds(), "len", len(allMsg))
 
-	c.addProgress((msgLen * 90) / total)
-	c.ConversationListener().OnSyncServerProgress(c.getProgress())
+	// log.ZDebug(ctx, "progress is", "msgLen", msgLen, "msgOffset", c.msgOffset, "total", total, "now progress is", (c.msgOffset*(100-InitSyncProgress))/total + InitSyncProgress)
+	c.ConversationListener().OnSyncServerProgress((c.msgOffset*(100-InitSyncProgress))/total + InitSyncProgress)
 }
 
-func (c *Conversation) addProgress(progress int) {
+func (c *Conversation) addInitProgress(progress int) {
 	c.progress += progress
 	if c.progress > 100 {
 		c.progress = 100
 	}
 }
 
-func (c *Conversation) getProgress() int {
-	return c.progress
-}
-
 func listToMap(list []*model_struct.LocalConversation, m map[string]*model_struct.LocalConversation) {
 	for _, v := range list {
 		m[v.ConversationID] = v

internal/conversation_msg/conversation_notification.go

@@ -40,6 +40,9 @@ const (
 	asyncWait
 )
 
+// InitSyncProgress is initialize Sync when reinstall.
+const InitSyncProgress = 10
+
 func (c *Conversation) Work(c2v common.Cmd2Value) {
 	log.ZDebug(c2v.Ctx, "NotificationCmd start", "cmd", c2v.Cmd, "value", c2v.Value)
 	defer log.ZDebug(c2v.Ctx, "NotificationCmd end", "cmd", c2v.Cmd, "value", c2v.Value)
@@ -65,28 +68,28 @@ func (c *Conversation) Work(c2v common.Cmd2Value) {
 func (c *Conversation) syncFlag(c2v common.Cmd2Value) {
 	ctx := c2v.Ctx
 	syncFlag := c2v.Value.(sdk_struct.CmdNewMsgComeToConversation).SyncFlag
-	initSyncBaseProgress := 5
 	switch syncFlag {
 	case constant.AppDataSyncStart:
 		log.ZDebug(ctx, "AppDataSyncStart")
 		c.startTime = time.Now()
 		c.ConversationListener().OnSyncServerStart(true)
+		c.ConversationListener().OnSyncServerProgress(1)
 		asyncWaitFunctions := []func(c context.Context) error{
 			c.group.SyncAllJoinedGroupsAndMembers,
 			c.friend.IncrSyncFriends,
 		}
 		runSyncFunctions(ctx, asyncWaitFunctions, asyncWait)
-		c.addProgress(initSyncBaseProgress)
-		c.ConversationListener().OnSyncServerProgress(c.getProgress()) // notify server current Progress
+		c.addInitProgress(InitSyncProgress * 4 / 10)              // add 40% of InitSyncProgress as progress
+		c.ConversationListener().OnSyncServerProgress(c.progress) // notify server current Progress
 
 		syncWaitFunctions := []func(c context.Context) error{
 			c.IncrSyncConversations,
 			c.SyncAllConversationHashReadSeqs,
 		}
 		runSyncFunctions(ctx, syncWaitFunctions, syncWait)
 		log.ZWarn(ctx, "core data sync over", nil, "cost time", time.Since(c.startTime).Seconds())
-		c.addProgress(initSyncBaseProgress)
-		c.ConversationListener().OnSyncServerProgress(c.getProgress()) // notify server current Progress
+		c.addInitProgress(InitSyncProgress * 6 / 10)              // add 60% of InitSyncProgress as progress
+		c.ConversationListener().OnSyncServerProgress(c.progress) // notify server current Progress
 
 		asyncNoWaitFunctions := []func(c context.Context) error{
 			c.user.SyncLoginUserInfoWithoutNotice,

internal/interaction/msg_sync.go

@@ -398,7 +398,11 @@ func (m *MsgSyncer) syncAndTriggerReinstallMsgs(ctx context.Context, seqMap map[
 				msgNum += int(oneConversationSyncNum)
 			}
 			if msgNum >= SplitPullMsgNum {
-				tpSeqMap := tempSeqMap
+				tpSeqMap := make(map[string][2]int64, len(tempSeqMap))
+				for k, v := range tempSeqMap {
+					tpSeqMap[k] = v
+				}
+
 				gr.Go(func() error {
 					resp, err := m.pullMsgBySeqRange(ctx, tpSeqMap, syncMsgNum)
 					if err != nil {