@@ -86,12 +86,12 @@ func (c *Conversation) checkEndBlock(ctx context.Context, conversationID string,
8686 // Perform an end-of-block check if the retrieved message count is less than requested
8787 if len (* list ) < count {
8888 if isReverse {
89- currentMaxSeq := c .maxSeqRecorder . Get ( conversationID )
89+ currentMaxSeq := c .getConversationMaxSeq ( ctx , conversationID )
9090 maxSeq , _ , _ := c .getMaxAndMinHaveSeqList (* list )
9191 log .ZDebug (ctx , "validateAndFillEndBlockContinuity" , "maxSeq" , maxSeq , "conversationID" , conversationID , "currentMaxSeq" , currentMaxSeq )
9292 // Use >= to prevent the currentMaxSeq from being updated too slowly,
9393 // which could lead to misjudgments and cause repeated message fetching."
94- if maxSeq >= currentMaxSeq { // todo Replace `1` with the minimum sequence value as defined by the user or system
94+ if maxSeq >= currentMaxSeq {
9595 messageListCallback .IsEnd = true
9696 } else {
9797 lastEndSeq , _ := c .messagePullReverseEndSeqMap .Load (conversationID )
@@ -104,7 +104,7 @@ func (c *Conversation) checkEndBlock(ctx context.Context, conversationID string,
104104 // The batch includes sequences but has not reached the maximum value,
105105 // This condition indicates local-only messages, with `maxSeq < maxSeqRecorderMaxSeq` as the only case,
106106 // since `lastEndSeq < maxSeqRecorderMaxSeq` is handled in inter-block continuity.
107- lostSeqList := getLostSeqListWithLimitLength (maxSeq + 1 , c . maxSeqRecorder . Get ( conversationID ) , []int64 {})
107+ lostSeqList := getLostSeqListWithLimitLength (maxSeq + 1 , currentMaxSeq , []int64 {})
108108 if len (lostSeqList ) > 0 {
109109 isShouldFetchMessage = true
110110 seqList = lostSeqList
@@ -115,22 +115,24 @@ func (c *Conversation) checkEndBlock(ctx context.Context, conversationID string,
115115 }
116116 return isShouldFetchMessage , seqList
117117 } else {
118+ userCanPullMinSeq := c .getConversationMinSeq (ctx , conversationID )
118119 _ , minSeq , _ := c .getMaxAndMinHaveSeqList (* list )
119- log .ZDebug (ctx , "validateAndFillEndBlockContinuity" , "minSeq" , minSeq , "conversationID" , conversationID )
120- if minSeq == 1 { // todo Replace `1` with the minimum sequence value as defined by the user or system
120+ log .ZDebug (ctx , "validateAndFillEndBlockContinuity" , "minSeq" , minSeq ,
121+ "conversationID" , conversationID , "userCanPullMinSeq" , userCanPullMinSeq )
122+ if minSeq == userCanPullMinSeq {
121123 messageListCallback .IsEnd = true
122124 } else {
123125 lastMinSeq , _ := c .messagePullForwardEndSeqMap .Load (conversationID )
124126 log .ZDebug (ctx , "validateAndFillEndBlockContinuity" , "lastMinSeq" , lastMinSeq , "conversationID" , conversationID )
125127 // If `minSeq` is zero and `lastMinSeq` is at the minimum server sequence, this batch is fully local
126- if minSeq == 0 && lastMinSeq == 1 { // All messages in this batch are local messages,
128+ if minSeq == 0 && lastMinSeq == userCanPullMinSeq { // All messages in this batch are local messages,
127129 // and the minimum seq of the last batch of valid messages has already reached the minimum pullable seq from the server.
128130 messageListCallback .IsEnd = true
129131 } else {
130132 // The batch includes sequences but has not reached the minimum value,
131133 // This condition indicates local-only messages, with `minSeq > 1` as the only case,
132134 // since `lastMinSeq > 1` is handled in inter-block continuity.
133- lostSeqList := getLostSeqListWithLimitLength (1 , minSeq - 1 , []int64 {})
135+ lostSeqList := getLostSeqListWithLimitLength (userCanPullMinSeq , minSeq - 1 , []int64 {})
134136 if len (lostSeqList ) > 0 {
135137 isShouldFetchMessage = true
136138 seqList = lostSeqList
@@ -200,6 +202,11 @@ func (c *Conversation) fetchAndMergeMissingMessages(ctx context.Context, convers
200202 conversationSeqs .ConversationID = conversationID
201203 conversationSeqs .Seqs = seqList
202204 getSeqMessageReq .Conversations = append (getSeqMessageReq .Conversations , & conversationSeqs )
205+ if isReverse {
206+ getSeqMessageReq .Order = sdkws .PullOrder_PullOrderAsc
207+ } else {
208+ getSeqMessageReq .Order = sdkws .PullOrder_PullOrderDesc
209+ }
203210 log .ZDebug (ctx , "conversation pull message, " , "req" , getSeqMessageReq )
204211 if startTime == 0 && ! c .LongConnMgr .IsConnected () {
205212 return
@@ -219,6 +226,9 @@ func (c *Conversation) fetchAndMergeMissingMessages(ctx context.Context, convers
219226 c .pullMessageIntoTable (ctx , getSeqMessageResp .Msgs )
220227 log .ZDebug (ctx , "syncMsgFromServerSplit pull msg success" ,
221228 "conversationID" , conversationID , "count" , count , "len" , len (* list ), "msgLen" , len (v .Msgs ))
229+ if v .IsEnd {
230+ c .setConversationMinSeq (ctx , isReverse , conversationID , v .EndSeq )
231+ }
222232 localMessage := datautil .Batch (MsgDataToLocalChatLog , v .Msgs )
223233 if ! isReverse {
224234 reverse (localMessage )
@@ -228,6 +238,56 @@ func (c *Conversation) fetchAndMergeMissingMessages(ctx context.Context, convers
228238
229239 }
230240}
241+
242+ func (c * Conversation ) getConversationMaxSeq (ctx context.Context , conversationID string ) int64 {
243+ conversation , err := c .db .GetConversation (ctx , conversationID )
244+ if err != nil {
245+ log .ZWarn (ctx , "Failed to get conversation" , err )
246+ return c .maxSeqRecorder .Get (conversationID )
247+ }
248+ if conversation .MaxSeq == 0 {
249+ return c .maxSeqRecorder .Get (conversationID )
250+
251+ }
252+ return conversation .MaxSeq
253+ }
254+ func (c * Conversation ) getConversationMinSeq (ctx context.Context , conversationID string ) int64 {
255+ conversation , err := c .db .GetConversation (ctx , conversationID )
256+ if err != nil {
257+ log .ZWarn (ctx , "Failed to get conversation" , err )
258+ return 1
259+ }
260+ if conversation .MinSeq == 0 {
261+ return 1
262+
263+ }
264+ return conversation .MinSeq
265+ }
266+ func (c * Conversation ) setConversationMinSeq (ctx context.Context , isReverse bool , conversationID string , endSeq int64 ) {
267+ conversation , err := c .db .GetConversation (ctx , conversationID )
268+ if err != nil {
269+ log .ZWarn (ctx , "Failed to get conversation" , err )
270+ return
271+ }
272+ if ! isReverse {
273+ if conversation .MinSeq == 0 || endSeq > conversation .MinSeq {
274+ conversation .MinSeq = endSeq
275+ }
276+ } else {
277+ if conversation .MaxSeq == 0 || endSeq < conversation .MaxSeq {
278+ conversation .MaxSeq = endSeq
279+ err = c .db .UpdateConversation (ctx , conversation )
280+ if err != nil {
281+ log .ZWarn (ctx , "Failed to update conversation" , err )
282+ }
283+ }
284+
285+ }
286+ err = c .db .UpdateConversation (ctx , conversation )
287+ if err != nil {
288+ log .ZWarn (ctx , "Failed to update conversation" , err )
289+ }
290+ }
231291func errHandle (seqList []int64 , list * []* model_struct.LocalChatLog , err error , messageListCallback * sdk.GetAdvancedHistoryMessageListCallback ) {
232292 messageListCallback .ErrCode = 100
233293 messageListCallback .ErrMsg = err .Error ()
0 commit comments