From 4457c4817390e5304482c8af5e0d3745455a66b5 Mon Sep 17 00:00:00 2001 From: highesttt Date: Sun, 28 Jun 2026 08:31:22 -0500 Subject: [PATCH] fix: backfill only worked for the 1st page of members --- pkg/connector/client.go | 6 +- pkg/connector/sync.go | 132 +++++++++++++++++++++++++++++++--------- pkg/line/structs.go | 11 ++-- 3 files changed, 115 insertions(+), 34 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 97f4795..eca68d1 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -35,7 +35,7 @@ type LineClient struct { recoverTime time.Time // cacheMu protects peerKeys, blockedUsers, contactCache, mediaFlowCache, - // noE2EEGroups, groupMemberCache, and generatedGroupNameCache. + // noE2EEGroups, groupMemberCache, generatedGroupNameCache, and knownMemberChatMIDs. // Hold it only around map accesses; never across network calls. cacheMu sync.Mutex blockedUsers map[string]bool // mid -> true if the user has blocked this contact in LINE @@ -44,6 +44,7 @@ type LineClient struct { mediaFlowCache map[string]cachedMediaFlow groupMemberCache map[string][]string // chatMid -> list of member MIDs from CreateGroup or getChatMemberMIDs generatedGroupNameCache map[string]bool // chatMid -> true when Matrix name should be generated from member names + knownMemberChatMIDs map[string]struct{} // chatMid -> current member chats returned by getAllChatMids reactionIconMXC map[int]string // predefinedReactionType -> cached MXC URI recentReactions sync.Map // "msgID\x00emoji" -> struct{} to dedup concurrent 139/140 events @@ -257,6 +258,9 @@ func (lc *LineClient) Connect(ctx context.Context) { if lc.groupMemberCache == nil { lc.groupMemberCache = make(map[string][]string) } + if lc.knownMemberChatMIDs == nil { + lc.knownMemberChatMIDs = make(map[string]struct{}) + } lc.cacheMu.Unlock() lc.reqSeqMu.Lock() if lc.sentReqSeqs == nil { diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 2bafdd7..4857877 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -27,11 +27,53 @@ import ( const ( prefetchMessagesConcurrency = 4 + messageBoxPageLimit = 100 + startupBackfillMessageLimit = 50 unblockBackfillFallbackDelay = 10 * time.Second groupPortalCreateWait = 30 * time.Second beeperExcludeFromTimelineKey = "com.beeper.exclude_from_timeline" ) +func (lc *LineClient) getMessageBoxesWithRecovery(ctx context.Context, opts line.MessageBoxesOptions) (*line.MessageBoxesResponse, error) { + client := lc.newClient() + res, err := client.GetMessageBoxes(opts) + if err != nil && (lc.isRefreshRequired(err) || lc.isLoggedOut(err)) { + if errRecover := lc.recoverToken(ctx); errRecover == nil { + client = lc.newClient() + res, err = client.GetMessageBoxes(opts) + } + } + return res, err +} + +func (lc *LineClient) fetchAllMessageBoxes(ctx context.Context, opts line.MessageBoxesOptions) ([]line.MessageBox, error) { + var boxes []line.MessageBox + seenCursors := make(map[string]struct{}) + for { + if err := ctx.Err(); err != nil { + return nil, err + } + res, err := lc.getMessageBoxesWithRecovery(ctx, opts) + if err != nil { + return nil, err + } + boxes = append(boxes, res.MessageBoxes...) + if !res.HasNext || len(res.MessageBoxes) == 0 { + return boxes, nil + } + + nextCursor := res.MessageBoxes[len(res.MessageBoxes)-1].ID + if nextCursor == "" { + return boxes, nil + } + if _, ok := seenCursors[nextCursor]; ok { + return boxes, nil + } + seenCursors[nextCursor] = struct{}{} + opts.MinChatID = nextCursor + } +} + func (lc *LineClient) refreshBlockedContacts(ctx context.Context) ([]string, error) { client := lc.newClient() blockedMIDs, err := client.GetBlockedContactIds() @@ -116,27 +158,20 @@ func (lc *LineClient) saveBlockedContactsSnapshot(ctx context.Context) { func (lc *LineClient) syncDMChats(ctx context.Context) { defer lc.wg.Done() - client := lc.newClient() opts := line.MessageBoxesOptions{ ActiveOnly: true, - MessageBoxCountLimit: 100, + MessageBoxCountLimit: messageBoxPageLimit, WithUnreadCount: false, LastMessagesPerMessageBoxCount: 0, } - res, err := client.GetMessageBoxes(opts) - if err != nil && (lc.isRefreshRequired(err) || lc.isLoggedOut(err)) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - res, err = client.GetMessageBoxes(opts) - } - } + messageBoxes, err := lc.fetchAllMessageBoxes(ctx, opts) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch message boxes for DM sync") return } - for _, box := range res.MessageBoxes { + for _, box := range messageBoxes { mid := box.ID lowerMid := strings.ToLower(mid) // Skip group chats — they're handled by syncChats @@ -331,36 +366,31 @@ func (lc *LineClient) FetchMessages(ctx context.Context, params bridgev2.FetchMe func (lc *LineClient) prefetchMessages(ctx context.Context) { defer lc.wg.Done() - client := lc.newClient() opts := line.MessageBoxesOptions{ ActiveOnly: true, - MessageBoxCountLimit: 100, + MessageBoxCountLimit: messageBoxPageLimit, WithUnreadCount: true, LastMessagesPerMessageBoxCount: 0, } - res, err := client.GetMessageBoxes(opts) - if err != nil && (lc.isRefreshRequired(err) || lc.isLoggedOut(err)) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - res, err = client.GetMessageBoxes(opts) - } - } + messageBoxes, err := lc.fetchAllMessageBoxes(ctx, opts) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to prefetch message boxes") return } + chatMIDs := collectStartupBackfillChatMIDs(messageBoxes, lc.getKnownMemberChatMIDs(), lc.isUserBlocked) workerCount := prefetchMessagesConcurrency - if len(res.MessageBoxes) < workerCount { - workerCount = len(res.MessageBoxes) + if len(chatMIDs) < workerCount { + workerCount = len(chatMIDs) } if workerCount == 0 { return } lc.UserLogin.Bridge.Log.Info(). - Int("chat_count", len(res.MessageBoxes)). + Int("message_box_count", len(messageBoxes)). + Int("chat_count", len(chatMIDs)). Int("concurrency", workerCount). Msg("Prefetching recent messages") @@ -374,27 +404,49 @@ func (lc *LineClient) prefetchMessages(ctx context.Context) { if ctx.Err() != nil { return } - lc.backfillRecentMessages(ctx, chatMID, 50) + lc.backfillRecentMessages(ctx, chatMID, startupBackfillMessageLimit) } }() } - for _, box := range res.MessageBoxes { - if lc.isUserBlocked(box.ID) { - continue - } + for _, chatMID := range chatMIDs { select { case <-ctx.Done(): close(jobs) workers.Wait() return - case jobs <- box.ID: + case jobs <- chatMID: } } close(jobs) workers.Wait() } +func collectStartupBackfillChatMIDs(messageBoxes []line.MessageBox, memberChatMIDs []string, isBlocked func(string) bool) []string { + seen := make(map[string]struct{}, len(messageBoxes)+len(memberChatMIDs)) + chatMIDs := make([]string, 0, len(messageBoxes)+len(memberChatMIDs)) + add := func(mid string) { + if mid == "" { + return + } + if isBlocked != nil && isBlocked(mid) { + return + } + if _, ok := seen[mid]; ok { + return + } + seen[mid] = struct{}{} + chatMIDs = append(chatMIDs, mid) + } + for _, box := range messageBoxes { + add(box.ID) + } + for _, mid := range memberChatMIDs { + add(mid) + } + return chatMIDs +} + // backfillRecentMessages fetches up to limit recent messages for a single // chat and queues any not already in the local DB through the normal inbound // (live) message path. Used by prefetchMessages on startup and as a delayed, @@ -463,8 +515,10 @@ func (lc *LineClient) syncChats(ctx context.Context) { allMids := append(midsResp.MemberChatMids, midsResp.InvitedChatMids...) if len(allMids) == 0 { + lc.setKnownMemberChatMIDs(nil) return } + lc.setKnownMemberChatMIDs(midsResp.MemberChatMids) memberChatMids := make(map[string]struct{}, len(midsResp.MemberChatMids)) for _, mid := range midsResp.MemberChatMids { memberChatMids[mid] = struct{}{} @@ -555,6 +609,28 @@ func (lc *LineClient) waitForGroupPortalCreates(ctx context.Context, portals []* } } +func (lc *LineClient) setKnownMemberChatMIDs(mids []string) { + lc.cacheMu.Lock() + defer lc.cacheMu.Unlock() + lc.knownMemberChatMIDs = make(map[string]struct{}, len(mids)) + for _, mid := range mids { + if isChatMID(mid) { + lc.knownMemberChatMIDs[mid] = struct{}{} + } + } +} + +func (lc *LineClient) getKnownMemberChatMIDs() []string { + lc.cacheMu.Lock() + defer lc.cacheMu.Unlock() + mids := make([]string, 0, len(lc.knownMemberChatMIDs)) + for mid := range lc.knownMemberChatMIDs { + mids = append(mids, mid) + } + sort.Strings(mids) + return mids +} + func (lc *LineClient) chatToChatInfo(ctx context.Context, chat *line.Chat, excludeFromTimeline bool) *bridgev2.ChatInfo { members := []bridgev2.ChatMember{ { diff --git a/pkg/line/structs.go b/pkg/line/structs.go index d81e1d7..4ae1d0a 100644 --- a/pkg/line/structs.go +++ b/pkg/line/structs.go @@ -273,11 +273,12 @@ type AcquireEncryptedAccessTokenResponse struct { } type MessageBoxesOptions struct { - ActiveOnly bool `json:"activeOnly"` - UnreadOnly bool `json:"unreadOnly"` - MessageBoxCountLimit int `json:"messageBoxCountLimit"` - WithUnreadCount bool `json:"withUnreadCount"` - LastMessagesPerMessageBoxCount int `json:"lastMessagesPerMessageBoxCount"` + MinChatID string `json:"minChatId,omitempty"` + ActiveOnly bool `json:"activeOnly"` + UnreadOnly bool `json:"unreadOnly"` + MessageBoxCountLimit int `json:"messageBoxCountLimit"` + WithUnreadCount bool `json:"withUnreadCount"` + LastMessagesPerMessageBoxCount int `json:"lastMessagesPerMessageBoxCount"` } type MessageBoxesResponse struct {