@@ -131,6 +131,100 @@ func (c *Conversation) SyncAllConversationHashReadSeqs(ctx context.Context) erro
131131 log .ZDebug (ctx , "TriggerCmdUpdateConversation completed" , "duration" , time .Since (stepStartTime ).Seconds ())
132132 }
133133
134+ stepStartTime = time .Now ()
135+ if err := c .syncAllGroupReadCursors (ctx ); err != nil {
136+ log .ZWarn (ctx , "syncAllGroupReadCursors failed" , err )
137+ }
138+ log .ZDebug (ctx , "syncAllGroupReadCursors completed" , "duration" , time .Since (stepStartTime ).Seconds ())
139+
134140 log .ZDebug (ctx , "SyncAllConversationHashReadSeqs completed" , "totalDuration" , time .Since (startTime ).Seconds ())
135141 return nil
136142}
143+
144+ func (c * Conversation ) syncAllGroupReadCursors (ctx context.Context ) error {
145+ // 1. 获取所有群聊会话ID
146+ conversations , err := c .db .GetAllConversations (ctx )
147+ if err != nil {
148+ log .ZWarn (ctx , "GetAllConversations failed" , err )
149+ return err
150+ }
151+
152+ var groupConversationIDs []string
153+ for _ , conv := range conversations {
154+ if conv .ConversationType == constant .ReadGroupChatType {
155+ groupConversationIDs = append (groupConversationIDs , conv .ConversationID )
156+ }
157+ }
158+
159+ if len (groupConversationIDs ) == 0 {
160+ log .ZDebug (ctx , "no group conversations to sync cursors" )
161+ return nil
162+ }
163+
164+ log .ZDebug (ctx , "found group conversations" , "count" , len (groupConversationIDs ), "conversationIDs" , groupConversationIDs )
165+
166+ // 2. 为每个群会话确保初始化 CursorState（如不存在则创建）
167+ for _ , conversationID := range groupConversationIDs {
168+ if _ , err := c .db .GetGroupReadCursorState (ctx , conversationID ); err != nil {
169+ if ierr := c .db .InsertGroupReadCursorState (ctx , & model_struct.LocalGroupReadCursorState {ConversationID : conversationID , CursorVersion : 1 }); ierr != nil {
170+ log .ZWarn (ctx , "InsertGroupReadCursorState failed" , ierr , "conversationID" , conversationID )
171+ } else {
172+ log .ZDebug (ctx , "initialized LocalGroupReadCursorState" , "conversationID" , conversationID )
173+ }
174+ }
175+ }
176+
177+ // 3. 调用服务器API获取已读游标
178+ stepStartTime := time .Now ()
179+ resp , err := c .getConversationReadCursors (ctx , groupConversationIDs )
180+ if err != nil {
181+ log .ZWarn (ctx , "getConversationReadCursorsFromServer failed" , err )
182+ return err
183+ }
184+ log .ZDebug (ctx , "getConversationReadCursorsFromServer completed" , "duration" , time .Since (stepStartTime ).Seconds ())
185+
186+ // 4. 处理响应并存储到本地数据库
187+ stepStartTime = time .Now ()
188+ allCursorCount := 0
189+ for conversationID , cursorList := range resp .Cursors {
190+ curCursorCount := 0
191+ if cursorList == nil || len (cursorList .Cursors ) == 0 {
192+ continue
193+ }
194+
195+ for _ , cursor := range cursorList .Cursors {
196+ localCursor := & model_struct.LocalGroupReadCursor {
197+ ConversationID : conversationID ,
198+ UserID : cursor .UserID ,
199+ MaxReadSeq : cursor .MaxReadSeq ,
200+ }
201+
202+ // 检查是否已存在，如果存在则更新，否则插入
203+ existingCursor , err := c .db .GetGroupReadCursor (ctx , conversationID , cursor .UserID )
204+ if err != nil {
205+ if err := c .db .InsertGroupReadCursor (ctx , localCursor ); err != nil {
206+ log .ZWarn (ctx , "InsertGroupReadCursor failed" , err , "conversationID" , conversationID , "userID" , cursor .UserID )
207+ } else {
208+ curCursorCount ++
209+ }
210+ } else {
211+ if cursor .MaxReadSeq > existingCursor .MaxReadSeq {
212+ if err := c .db .UpdateGroupReadCursor (ctx , conversationID , cursor .UserID , cursor .MaxReadSeq ); err != nil {
213+ log .ZWarn (ctx , "UpdateGroupReadCursor failed" , err , "conversationID" , conversationID , "userID" , cursor .UserID )
214+ } else {
215+ curCursorCount ++
216+ }
217+ }
218+ }
219+ }
220+ allCursorCount += curCursorCount
221+ if curCursorCount != 0 {
222+ if err := c .db .IncrementGroupReadCursorVersion (ctx , conversationID ); err != nil {
223+ log .ZWarn (ctx , "IncrementGroupReadCursorVersion failed" , err , "conversationID" , conversationID )
224+ }
225+ }
226+ }
227+
228+ log .ZDebug (ctx , "syncAllGroupReadCursors completed" , "duration" , time .Since (stepStartTime ).Seconds (), "cursorCount" , allCursorCount )
229+ return nil
230+ }
0 commit comments