From f8e0d682f4938ad87edb25ab32d56760fb78c4a6 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 11 Jun 2026 18:25:50 -0500 Subject: [PATCH 1/4] fix: backfill not working in specific circumstances --- pkg/connector/client.go | 2 +- pkg/connector/sync.go | 90 ++++++++++++++++++++++++++++++++++++----- 2 files changed, 81 insertions(+), 11 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 47df04e..5f723be 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -278,7 +278,7 @@ func (lc *LineClient) Connect(ctx context.Context) { } // Fetch initial blocked contacts list before starting sync loops. - if err := lc.refreshBlockedContacts(ctx); err != nil { + if _, err := lc.refreshBlockedContacts(ctx); err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch blocked contacts, continuing without block list") } diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 03d730b..8c2fa54 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -24,9 +24,12 @@ import ( "github.com/highesttt/matrix-line-messenger/pkg/line" ) -const prefetchMessagesConcurrency = 4 +const ( + prefetchMessagesConcurrency = 4 + unblockBackfillFallbackDelay = 10 * time.Second +) -func (lc *LineClient) refreshBlockedContacts(ctx context.Context) error { +func (lc *LineClient) refreshBlockedContacts(ctx context.Context) ([]string, error) { client := line.NewClient(lc.AccessToken) blockedMIDs, err := client.GetBlockedContactIds() if err != nil && (lc.isRefreshRequired(err) || lc.isLoggedOut(err)) { @@ -36,7 +39,7 @@ func (lc *LineClient) refreshBlockedContacts(ctx context.Context) error { } } if err != nil { - return err + return nil, err } blockedUsers := make(map[string]bool, len(blockedMIDs)) @@ -45,11 +48,17 @@ func (lc *LineClient) refreshBlockedContacts(ctx context.Context) error { } lc.cacheMu.Lock() + var newlyUnblocked []string + for mid := range lc.blockedUsers { + if !blockedUsers[mid] { + newlyUnblocked = append(newlyUnblocked, mid) + } + } lc.blockedUsers = blockedUsers lc.cacheMu.Unlock() lc.UserLogin.Bridge.Log.Info().Int("count", len(blockedMIDs)).Msg("Refreshed blocked contacts") - return nil + return newlyUnblocked, nil } func (lc *LineClient) syncDMChats(ctx context.Context) { @@ -147,6 +156,40 @@ func (lc *LineClient) queueDMChatResync(ctx context.Context, mid string, createP }) } +func (lc *LineClient) queueUnblockBackfillFallback(ctx context.Context, mid string) { + lc.wg.Add(1) + go func() { + defer lc.wg.Done() + + timer := time.NewTimer(unblockBackfillFallbackDelay) + defer timer.Stop() + + select { + case <-ctx.Done(): + return + case <-timer.C: + } + + if lc.isUserBlocked(mid) { + lc.UserLogin.Bridge.Log.Debug(). + Str("mid", mid). + Msg("Skipping unblock fallback backfill because contact is blocked again") + return + } + + lc.UserLogin.Bridge.Log.Info(). + Str("mid", mid). + Dur("delay", unblockBackfillFallbackDelay). + Msg("Running unblock fallback recent-message backfill") + lc.backfillRecentMessages(ctx, mid, 50) + }() +} + +func isLineGroupOrRoomMID(mid string) bool { + lowerMid := strings.ToLower(mid) + return strings.HasPrefix(lowerMid, "c") || strings.HasPrefix(lowerMid, "r") +} + // FetchMessages implements bridgev2.BackfillingNetworkAPI. It powers silent, // batch-sent history backfill. It is currently triggered when a DM portal is // recreated after the contact is unblocked (see queueDMChatResync), repopulating @@ -297,16 +340,25 @@ func (lc *LineClient) prefetchMessages(ctx context.Context) { // 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. Note that this -// notifies for any not-yet-bridged messages; the silent backfill path used on -// unblock goes through FetchMessages instead. +// (live) message path. Used by prefetchMessages on startup and as a delayed, +// deduped fallback after unblock if bridgev2's silent forward backfill didn't +// populate the restored room first. func (lc *LineClient) backfillRecentMessages(ctx context.Context, chatMID string, limit int) { + start := time.Now() client := line.NewClient(lc.AccessToken) msgs, err := client.GetRecentMessagesV2(chatMID, limit) + if err != nil && (lc.isRefreshRequired(err) || lc.isLoggedOut(err)) { + if errRecover := lc.recoverToken(ctx); errRecover == nil { + client = line.NewClient(lc.AccessToken) + msgs, err = client.GetRecentMessagesV2(chatMID, limit) + } + } if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chatMID).Msg("Failed to fetch recent messages") return } + queued := 0 + skippedExisting := 0 // Reverse messages to process oldest first for i := len(msgs) - 1; i >= 0; i-- { msg := msgs[i] @@ -316,6 +368,7 @@ func (lc *LineClient) backfillRecentMessages(ctx context.Context, chatMID string existing, err := lc.UserLogin.Bridge.DB.Message.GetPartByID(ctx, lc.UserLogin.ID, networkid.MessageID(msg.ID), "") if err == nil && existing != nil { + skippedExisting++ continue } @@ -324,7 +377,15 @@ func (lc *LineClient) backfillRecentMessages(ctx context.Context, chatMID string opType = OpSendMessage } lc.queueIncomingMessage(msg, int(opType)) + queued++ } + lc.UserLogin.Bridge.Log.Debug(). + Str("chat_mid", chatMID). + Int("fetched", len(msgs)). + Int("queued", queued). + Int("skipped_existing", skippedExisting). + Dur("duration", time.Since(start)). + Msg("Finished recent-message backfill") } func (lc *LineClient) syncChats(ctx context.Context) { @@ -761,9 +822,18 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } } - if err := lc.refreshBlockedContacts(ctx); err != nil { + newlyUnblocked, err := lc.refreshBlockedContacts(ctx) + if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to refresh blocked contacts during fullSync") } + for _, mid := range newlyUnblocked { + if isLineGroupOrRoomMID(mid) { + continue + } + lc.UserLogin.Bridge.Log.Info().Str("mid", mid).Msg("Detected unblocked contact during fullSync") + lc.queueDMChatResync(ctx, mid, true, true) + lc.queueUnblockBackfillFallback(ctx, mid) + } lc.wg.Add(3) go lc.syncChats(ctx) go lc.syncDMChats(ctx) @@ -875,11 +945,11 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { // restored chat repopulates without firing a // notification for every old message — a blocked contact can't have sent // anything new, so notifying on unblock is never useful. - lowerMid := strings.ToLower(mid) - if strings.HasPrefix(lowerMid, "c") || strings.HasPrefix(lowerMid, "r") { + if isLineGroupOrRoomMID(mid) { return } lc.queueDMChatResync(ctx, mid, true, true) + lc.queueUnblockBackfillFallback(ctx, mid) case OpContactUpdate: mid := op.Param1 From 34714d178add53a94bb6733bf550f3cfd9b48673 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 11 Jun 2026 18:50:53 -0500 Subject: [PATCH 2/4] chore: indent fixes - unblocks while bridge was offline were not detected - unblocks could trigger unwanted notifications - removed duplicate function --- pkg/connector/client.go | 6 ++- pkg/connector/connector.go | 1 + pkg/connector/sync.go | 84 ++++++++++++++++++++++++++++++-------- 3 files changed, 73 insertions(+), 18 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 5f723be..e552d91 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -278,8 +278,12 @@ func (lc *LineClient) Connect(ctx context.Context) { } // Fetch initial blocked contacts list before starting sync loops. - if _, err := lc.refreshBlockedContacts(ctx); err != nil { + if newlyUnblocked, err := lc.refreshBlockedContacts(ctx); err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch blocked contacts, continuing without block list") + } else { + for _, mid := range newlyUnblocked { + lc.queueUnblockedDMRestore(ctx, mid, "startup") + } } lc.wg.Add(4) diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 4a9f213..a8a0764 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -110,6 +110,7 @@ type UserLoginMetadata struct { E2EEVersion string `json:"e2ee_version,omitempty"` E2EEKeyID string `json:"e2ee_key_id,omitempty"` ExportedKeyMap map[string]string `json:"exported_key_map,omitempty"` + BlockedMIDs []string `json:"blocked_mids,omitempty"` } func (lc *LineConnector) LoadUserLogin(ctx context.Context, login *bridgev2.UserLogin) error { diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 8c2fa54..b87ad5e 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "sort" "strconv" "strings" "sync" @@ -47,9 +48,17 @@ func (lc *LineClient) refreshBlockedContacts(ctx context.Context) ([]string, err blockedUsers[mid] = true } + metadataBlockedUsers := lc.metadataBlockedContacts() lc.cacheMu.Lock() - var newlyUnblocked []string + previousBlockedUsers := make(map[string]bool, len(lc.blockedUsers)+len(metadataBlockedUsers)) for mid := range lc.blockedUsers { + previousBlockedUsers[mid] = true + } + for mid := range metadataBlockedUsers { + previousBlockedUsers[mid] = true + } + var newlyUnblocked []string + for mid := range previousBlockedUsers { if !blockedUsers[mid] { newlyUnblocked = append(newlyUnblocked, mid) } @@ -57,10 +66,51 @@ func (lc *LineClient) refreshBlockedContacts(ctx context.Context) ([]string, err lc.blockedUsers = blockedUsers lc.cacheMu.Unlock() + lc.saveBlockedContacts(ctx, blockedUsers) lc.UserLogin.Bridge.Log.Info().Int("count", len(blockedMIDs)).Msg("Refreshed blocked contacts") return newlyUnblocked, nil } +func (lc *LineClient) metadataBlockedContacts() map[string]bool { + blockedUsers := make(map[string]bool) + meta, ok := lc.UserLogin.Metadata.(*UserLoginMetadata) + if !ok { + return blockedUsers + } + for _, mid := range meta.BlockedMIDs { + blockedUsers[mid] = true + } + return blockedUsers +} + +func (lc *LineClient) saveBlockedContacts(ctx context.Context, blockedUsers map[string]bool) { + meta, ok := lc.UserLogin.Metadata.(*UserLoginMetadata) + if !ok { + return + } + + mids := make([]string, 0, len(blockedUsers)) + for mid := range blockedUsers { + mids = append(mids, mid) + } + sort.Strings(mids) + meta.BlockedMIDs = mids + if err := lc.UserLogin.Save(ctx); err != nil { + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to save blocked contacts snapshot") + } +} + +func (lc *LineClient) saveBlockedContactsSnapshot(ctx context.Context) { + lc.cacheMu.Lock() + blockedUsers := make(map[string]bool, len(lc.blockedUsers)) + for mid := range lc.blockedUsers { + blockedUsers[mid] = true + } + lc.cacheMu.Unlock() + + lc.saveBlockedContacts(ctx, blockedUsers) +} + func (lc *LineClient) syncDMChats(ctx context.Context) { defer lc.wg.Done() @@ -180,14 +230,21 @@ func (lc *LineClient) queueUnblockBackfillFallback(ctx context.Context, mid stri lc.UserLogin.Bridge.Log.Info(). Str("mid", mid). Dur("delay", unblockBackfillFallbackDelay). - Msg("Running unblock fallback recent-message backfill") - lc.backfillRecentMessages(ctx, mid, 50) + Msg("Re-requesting silent unblock backfill") + lc.queueDMChatResync(ctx, mid, true, true) }() } -func isLineGroupOrRoomMID(mid string) bool { - lowerMid := strings.ToLower(mid) - return strings.HasPrefix(lowerMid, "c") || strings.HasPrefix(lowerMid, "r") +func (lc *LineClient) queueUnblockedDMRestore(ctx context.Context, mid, reason string) { + if isChatMID(mid) { + return + } + lc.UserLogin.Bridge.Log.Info(). + Str("mid", mid). + Str("reason", reason). + Msg("Restoring unblocked DM with backfill") + lc.queueDMChatResync(ctx, mid, true, true) + lc.queueUnblockBackfillFallback(ctx, mid) } // FetchMessages implements bridgev2.BackfillingNetworkAPI. It powers silent, @@ -827,12 +884,7 @@ func (lc *LineClient) pollLoop(ctx context.Context) { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to refresh blocked contacts during fullSync") } for _, mid := range newlyUnblocked { - if isLineGroupOrRoomMID(mid) { - continue - } - lc.UserLogin.Bridge.Log.Info().Str("mid", mid).Msg("Detected unblocked contact during fullSync") - lc.queueDMChatResync(ctx, mid, true, true) - lc.queueUnblockBackfillFallback(ctx, mid) + lc.queueUnblockedDMRestore(ctx, mid, "full_sync") } lc.wg.Add(3) go lc.syncChats(ctx) @@ -916,6 +968,7 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { lc.cacheMu.Lock() lc.blockedUsers[mid] = true lc.cacheMu.Unlock() + lc.saveBlockedContactsSnapshot(ctx) lc.UserLogin.Bridge.Log.Info().Str("mid", mid).Msg("Contact blocked") // Block operations should only carry user MIDs; skip if it looks like a group/room // to avoid blast-radius deleting a group portal on an unexpected payload. @@ -939,17 +992,14 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { lc.cacheMu.Lock() delete(lc.blockedUsers, mid) lc.cacheMu.Unlock() + lc.saveBlockedContactsSnapshot(ctx) lc.UserLogin.Bridge.Log.Info().Str("mid", mid).Msg("Contact unblocked") // Reattach the DM portal and request an immediate backfill in the same // resync. The backfill is batch-sent silently (see FetchMessages), so the // restored chat repopulates without firing a // notification for every old message — a blocked contact can't have sent // anything new, so notifying on unblock is never useful. - if isLineGroupOrRoomMID(mid) { - return - } - lc.queueDMChatResync(ctx, mid, true, true) - lc.queueUnblockBackfillFallback(ctx, mid) + lc.queueUnblockedDMRestore(ctx, mid, "op_unblock") case OpContactUpdate: mid := op.Param1 From 31d0ed0887cba3563b738dea5c6b616b9325c3f6 Mon Sep 17 00:00:00 2001 From: Highest <91629626+highesttt@users.noreply.github.com> Date: Thu, 11 Jun 2026 18:58:20 -0500 Subject: [PATCH 3/4] Update pkg/connector/sync.go Co-authored-by: indent[bot] <216979840+indent[bot]@users.noreply.github.com> --- pkg/connector/sync.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index b87ad5e..9768215 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -399,7 +399,9 @@ func (lc *LineClient) prefetchMessages(ctx context.Context) { // 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, // deduped fallback after unblock if bridgev2's silent forward backfill didn't -// populate the restored room first. +// 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. func (lc *LineClient) backfillRecentMessages(ctx context.Context, chatMID string, limit int) { start := time.Now() client := line.NewClient(lc.AccessToken) From 6601c1842dc43b84182fa74ef92c624b0521d76d Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 11 Jun 2026 19:03:08 -0500 Subject: [PATCH 4/4] chore: failsafe for unblocks after bridge restart --- pkg/connector/client.go | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index e552d91..2221840 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -277,9 +277,17 @@ func (lc *LineClient) Connect(ctx context.Context) { } } + // Seed the last-known block list before fetching a fresh copy, so a + // transient LINE API failure doesn't reopen intentionally blocked DMs. + lc.cacheMu.Lock() + for mid := range lc.metadataBlockedContacts() { + lc.blockedUsers[mid] = true + } + lc.cacheMu.Unlock() + // Fetch initial blocked contacts list before starting sync loops. if newlyUnblocked, err := lc.refreshBlockedContacts(ctx); err != nil { - lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch blocked contacts, continuing without block list") + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch blocked contacts, continuing with last known block list") } else { for _, mid := range newlyUnblocked { lc.queueUnblockedDMRestore(ctx, mid, "startup")