From c63981be768f02aa11abc2505a7a6197de12bd38 Mon Sep 17 00:00:00 2001 From: Indent Date: Sun, 8 Nov 2026 06:27:35 +0000 Subject: [PATCH 1/4] backfill: don't requeue history sync conversations that are still being created --- pkg/connector/backfill.go | 9 +++++++++ pkg/connector/client.go | 2 ++ 2 files changed, 11 insertions(+) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index cbc7eec..2d33535 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -442,6 +442,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { // We don't currently support new PSAs, so don't bother backfilling them either wg.Done() continue + } else if wa.historySyncQueued.Has(conv.ChatJID) { + wg.Done() + continue } // TODO can the chat info fetch be avoided entirely? select { @@ -482,6 +485,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } + wa.historySyncQueued.Add(conv.ChatJID) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -497,6 +501,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { if err != nil { zerolog.Ctx(ctx).Err(err).Msg("Failed to mark conversation as bridged") } + wa.historySyncQueued.Remove(conv.ChatJID) wg.Done() }, }, @@ -504,6 +509,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { + wa.historySyncQueued.Remove(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -511,6 +517,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log.Info().Int("conversation_count", len(conversations)).Msg("Finished creating portals from history sync") go func() { wg.Wait() + if wa.historySyncQueued.Size() > 0 { + return + } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false err = wa.UserLogin.Save(ctx) if err != nil { diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 678c8d2..3630972 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -50,6 +50,7 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2. MC: noopMCInstance, historySyncWakeup: make(chan struct{}, 1), + historySyncQueued: exsync.NewSet[types.JID](), resyncQueue: make(map[types.JID]resyncQueueItem), directMediaRetries: make(map[networkid.MessageID]*directMediaRetry), mediaRetryLock: semaphore.NewWeighted(wa.Config.HistorySync.MediaRequests.MaxAsyncHandle), @@ -114,6 +115,7 @@ type WhatsAppClient struct { MC mClient historySyncWakeup chan struct{} + historySyncQueued *exsync.Set[types.JID] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex From efdaad906581d8db6a24dd0a6c19a191711f5bbe Mon Sep 17 00:00:00 2001 From: Indent Date: Sun, 8 Nov 2026 07:23:11 +0000 Subject: [PATCH 2/4] backfill: let in-flight history sync entries expire so failed creations are retried --- pkg/connector/backfill.go | 12 +++++++----- pkg/connector/client.go | 4 ++-- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index 2d33535..e2a18b1 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -399,6 +399,8 @@ func (wa *WhatsAppClient) handleWAHistorySync( return nil } +const historySyncQueuedTimeout = 10 * time.Minute + func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log := wa.UserLogin.Log.With(). Str("action", "create portals from history sync"). @@ -442,7 +444,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { // We don't currently support new PSAs, so don't bother backfilling them either wg.Done() continue - } else if wa.historySyncQueued.Has(conv.ChatJID) { + } else if queuedAt, ok := wa.historySyncQueued.Get(conv.ChatJID); ok && time.Since(queuedAt) < historySyncQueuedTimeout { wg.Done() continue } @@ -485,7 +487,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } - wa.historySyncQueued.Add(conv.ChatJID) + wa.historySyncQueued.Set(conv.ChatJID, time.Now()) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -501,7 +503,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { if err != nil { zerolog.Ctx(ctx).Err(err).Msg("Failed to mark conversation as bridged") } - wa.historySyncQueued.Remove(conv.ChatJID) + wa.historySyncQueued.Delete(conv.ChatJID) wg.Done() }, }, @@ -509,7 +511,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { - wa.historySyncQueued.Remove(conv.ChatJID) + wa.historySyncQueued.Delete(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -517,7 +519,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log.Info().Int("conversation_count", len(conversations)).Msg("Finished creating portals from history sync") go func() { wg.Wait() - if wa.historySyncQueued.Size() > 0 { + if wa.historySyncQueued.Len() > 0 { return } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 3630972..9b3f190 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -50,7 +50,7 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2. MC: noopMCInstance, historySyncWakeup: make(chan struct{}, 1), - historySyncQueued: exsync.NewSet[types.JID](), + historySyncQueued: exsync.NewMap[types.JID, time.Time](), resyncQueue: make(map[types.JID]resyncQueueItem), directMediaRetries: make(map[networkid.MessageID]*directMediaRetry), mediaRetryLock: semaphore.NewWeighted(wa.Config.HistorySync.MediaRequests.MaxAsyncHandle), @@ -115,7 +115,7 @@ type WhatsAppClient struct { MC mClient historySyncWakeup chan struct{} - historySyncQueued *exsync.Set[types.JID] + historySyncQueued *exsync.Map[types.JID, time.Time] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex From 5c8c1fc8358c011891fbf6066f289b9ad5fee223 Mon Sep 17 00:00:00 2001 From: "indent[bot]" <297468268+indent[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:09:28 +0000 Subject: [PATCH 3/4] handlewhatsapp: ignore HD replacement messages without replacement content (#1037) Co-authored-by: Indent --- pkg/connector/handlewhatsapp.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/pkg/connector/handlewhatsapp.go b/pkg/connector/handlewhatsapp.go index 4391233..e6174b9 100644 --- a/pkg/connector/handlewhatsapp.go +++ b/pkg/connector/handlewhatsapp.go @@ -374,6 +374,14 @@ func (wa *WhatsAppClient) handleWAMessage(ctx context.Context, evt *events.Messa if assocType := messageAssoc.GetAssociationType(); assocType == waE2E.MessageAssociation_HD_IMAGE_DUAL_UPLOAD || assocType == waE2E.MessageAssociation_HD_VIDEO_DUAL_UPLOAD { parentKey := messageAssoc.GetParentMessageKey() protocolMsg, shouldHideEdit := makeHDMediaReplacementEdit(evt.Message, parentKey) + if protocolMsg.GetEditedMessage() == nil { + wa.UserLogin.Log.Debug(). + Str("message_id", evt.Info.ID). + Str("parent_id", parentKey.GetID()). + Stringer("assoc_type", assocType). + Msg("Ignoring HD replacement message without replacement content") + return + } dontRenderEdited = shouldHideEdit wa.UserLogin.Log.Debug(). Str("message_id", evt.Info.ID). From 945474343fcd7579c3d273acb78f6c60c957e2e0 Mon Sep 17 00:00:00 2001 From: "indent[bot]" <297468268+indent[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:09:56 +0000 Subject: [PATCH 4/4] backfill: log stub and message type of unparseable history messages (#1034) Co-authored-by: Indent --- pkg/connector/backfill.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index cbc7eec..8f006e0 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -298,6 +298,9 @@ func (wa *WhatsAppClient) handleWAHistorySync( Int("msg_index", i). Str("msg_id", rawMsg.GetMessage().GetKey().GetID()). Uint64("msg_time_seconds", rawMsg.GetMessage().GetMessageTimestamp()). + Stringer("stub_type", rawMsg.GetMessage().GetMessageStubType()). + Str("message_type", getMessageType(rawMsg.GetMessage().GetMessage())). + Bool("from_me", rawMsg.GetMessage().GetKey().GetFromMe()). Msg("Dropping historical message due to parse error") log.Trace(). Any("web_message_info", rawMsg.GetMessage()).