Skip to content

fix: backfill only worked for the 1st page of members #194

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
highesttt merged 1 commit into from
Jun 28, 2026
+115 −34
Merged

fix: backfill only worked for the 1st page of members #194

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion pkg/connector/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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 {
Expand Down
132 changes: 104 additions & 28 deletions pkg/connector/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment on lines +65 to +70

@coderabbitai coderabbitai Bot Jun 28, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Don't treat a broken pagination cursor as success.

If HasNext is still true but the API returns an empty or repeated ID, this helper returns boxes, nil and the callers proceed with a silently truncated sync/backfill result. That recreates the “first page only” failure mode without any signal.

Suggested fix 
 		nextCursor := res.MessageBoxes[len(res.MessageBoxes)-1].ID
 		if nextCursor == "" {
-			return boxes, nil
+			return nil, fmt.Errorf("message-box pagination returned empty cursor with hasNext=true")
 		}
 		if _, ok := seenCursors[nextCursor]; ok {
-			return boxes, nil
+			return nil, fmt.Errorf("message-box pagination cursor %q repeated", nextCursor)
 		}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
nextCursor := res.MessageBoxes[len(res.MessageBoxes)-1].ID
if nextCursor == "" {
return boxes, nil
}
if _, ok := seenCursors[nextCursor]; ok {
return boxes, nil
nextCursor := res.MessageBoxes[len(res.MessageBoxes)-1].ID
if nextCursor == "" {
return nil, fmt.Errorf("message-box pagination returned empty cursor with hasNext=true")
}
if _, ok := seenCursors[nextCursor]; ok {
return nil, fmt.Errorf("message-box pagination cursor %q repeated", nextCursor)
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/connector/sync.go` around lines 65 - 70, The pagination logic in
sync/backfill helpers is treating an empty or repeated cursor as a successful
end-of-data case, which can silently truncate results. Update the cursor
handling in the helper that reads res.MessageBoxes and tracks seenCursors so
that when HasNext is true but nextCursor is empty or already seen, it returns an
error instead of boxes, nil. Make sure callers can distinguish a broken cursor
from a normal completion by surfacing a clear failure from this pagination path.

}
seenCursors[nextCursor] = struct{}{}
opts.MinChatID = nextCursor
}
}

func (lc *LineClient) refreshBlockedContacts(ctx context.Context) ([]string, error) {
client := lc.newClient()
blockedMIDs, err := client.GetBlockedContactIds()
Expand Down Expand Up @@ -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 {

@indent-zero indent-zero Bot Jun 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Latent: duplicate ChatResync on overlapping pages. fetchAllMessageBoxes uses the last box's ID as the next MinChatID; if LINE treats minChatId as inclusive (a common cursor convention — the protocol isn't documented either way), each page will repeat the previous page's last box, and syncDMChats will emit ChatResync twice for that DM. seenCursors only stops the loop when the cursor doesn't advance at all, so overlap-by-one passes through. The framework should treat ChatResync as idempotent so this is benign today, but prefetchMessages already dedupes via collectStartupBackfillChatMIDs — consider doing the same here for symmetry, or dropping the last box from each non-final page in fetchAllMessageBoxes.

mid := box.ID
lowerMid := strings.ToLower(mid)
// Skip group chats — they're handled by syncChats
Expand Down Expand Up @@ -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")

Expand All @@ -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,
Expand Down Expand Up @@ -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{}{}
Expand Down Expand Up @@ -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{
{
Expand Down
11 changes: 6 additions & 5 deletions pkg/line/structs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Loading