From 01243d430d5e2eab3117bdf5db17c644b0c8517a Mon Sep 17 00:00:00 2001 From: Indent Date: Wed, 11 Nov 2026 06:16:58 +0000 Subject: [PATCH 1/5] backfill: re-arm history sync portal creation after new history sync payloads --- pkg/connector/backfill.go | 12 ++++++++++++ pkg/connector/client.go | 2 ++ 2 files changed, 14 insertions(+) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index e2c7d114..c9cc4037 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -96,6 +96,14 @@ func (wa *WhatsAppClient) historySyncLoop(ctx context.Context) { } if resetTimer { timerPending.Store(true) + wa.historySyncPortalGen.Add(1) + if meta := wa.UserLogin.Metadata.(*waid.UserLoginMetadata); !meta.HistorySyncPortalsNeedCreating { + meta.HistorySyncPortalsNeedCreating = true + err := wa.UserLogin.Save(ctx) + if err != nil { + wa.UserLogin.Log.Err(err).Msg("Failed to save user login history sync portals need creating flag") + } + } } if timerPending.Load() { dispatchTimer.Reset(wa.Main.Config.HistorySync.DispatchWait) @@ -417,6 +425,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { } }() ctx = log.WithContext(ctx) + gen := wa.historySyncPortalGen.Load() limit := wa.Main.Config.HistorySync.MaxInitialConversations loginTS := wa.UserLogin.Metadata.(*waid.UserLoginMetadata).LoggedInAt conversations, err := wa.Main.DB.Conversation.GetRecent(ctx, wa.UserLogin.ID, limit, loginTS) @@ -517,6 +526,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.historySyncPortalGen.Load() != gen { + 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 678c8d22..62f9a461 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -128,6 +128,8 @@ type WhatsAppClient struct { lastPresence types.Presence blocklistLock sync.Mutex + historySyncPortalGen atomic.Uint64 + disableNewsletter bool appStateRecoveryLock sync.Mutex From 664956a43b8d32388d4287916b4c5b4e151f644d Mon Sep 17 00:00:00 2001 From: Indent Date: Wed, 11 Nov 2026 06:58:41 +0000 Subject: [PATCH 2/5] backfill: serialize history sync portal flag updates --- pkg/connector/backfill.go | 12 +++++++++--- pkg/connector/client.go | 3 ++- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index c9cc4037..27b23a5b 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -96,7 +96,8 @@ func (wa *WhatsAppClient) historySyncLoop(ctx context.Context) { } if resetTimer { timerPending.Store(true) - wa.historySyncPortalGen.Add(1) + wa.historySyncPortalLock.Lock() + wa.historySyncPortalGen++ if meta := wa.UserLogin.Metadata.(*waid.UserLoginMetadata); !meta.HistorySyncPortalsNeedCreating { meta.HistorySyncPortalsNeedCreating = true err := wa.UserLogin.Save(ctx) @@ -104,6 +105,7 @@ func (wa *WhatsAppClient) historySyncLoop(ctx context.Context) { wa.UserLogin.Log.Err(err).Msg("Failed to save user login history sync portals need creating flag") } } + wa.historySyncPortalLock.Unlock() } if timerPending.Load() { dispatchTimer.Reset(wa.Main.Config.HistorySync.DispatchWait) @@ -425,7 +427,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { } }() ctx = log.WithContext(ctx) - gen := wa.historySyncPortalGen.Load() + wa.historySyncPortalLock.Lock() + gen := wa.historySyncPortalGen + wa.historySyncPortalLock.Unlock() limit := wa.Main.Config.HistorySync.MaxInitialConversations loginTS := wa.UserLogin.Metadata.(*waid.UserLoginMetadata).LoggedInAt conversations, err := wa.Main.DB.Conversation.GetRecent(ctx, wa.UserLogin.ID, limit, loginTS) @@ -526,7 +530,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.historySyncPortalGen.Load() != gen { + wa.historySyncPortalLock.Lock() + defer wa.historySyncPortalLock.Unlock() + if wa.historySyncPortalGen != gen { return } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 62f9a461..cfb48708 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -128,7 +128,8 @@ type WhatsAppClient struct { lastPresence types.Presence blocklistLock sync.Mutex - historySyncPortalGen atomic.Uint64 + historySyncPortalLock sync.Mutex + historySyncPortalGen uint64 disableNewsletter bool From 7851e95028c35f5dd9776c3bfd90ac9f4964f4f1 Mon Sep 17 00:00:00 2001 From: "indent[bot]" <297468268+indent[bot]@users.noreply.github.com> Date: Thu, 12 Nov 2026 06:06:29 -0700 Subject: [PATCH 3/5] msgconv: don't look up LID mentions without a client (#1050) Co-authored-by: Indent --- pkg/msgconv/from-whatsapp.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pkg/msgconv/from-whatsapp.go b/pkg/msgconv/from-whatsapp.go index ec273076..e2ab92cd 100644 --- a/pkg/msgconv/from-whatsapp.go +++ b/pkg/msgconv/from-whatsapp.go @@ -76,8 +76,7 @@ func (mc *MessageConverter) getBasicUserInfo(ctx context.Context, user types.JID var pnJID types.JID if user.Server == types.DefaultUserServer { pnJID = user - } else if user.Server == types.HiddenUserServer { - cli := getClient(ctx) + } else if cli := getClient(ctx); user.Server == types.HiddenUserServer && cli != nil { if user.User == cli.Store.GetLID().User { pnJID = cli.Store.GetJID() } else { From 37055ae4fcc8d5541ab9819f8bea94095d33d862 Mon Sep 17 00:00:00 2001 From: "indent[bot]" <297468268+indent[bot]@users.noreply.github.com> Date: Thu, 12 Nov 2026 06:07:31 -0700 Subject: [PATCH 4/5] wadb: index history sync messages by chat and timestamp (#1051) Co-authored-by: Indent --- pkg/connector/wadb/upgrades/00-latest-schema.sql | 4 +++- .../wadb/upgrades/12-history-sync-message-timestamp-index.sql | 2 ++ 2 files changed, 5 insertions(+), 1 deletion(-) create mode 100644 pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql diff --git a/pkg/connector/wadb/upgrades/00-latest-schema.sql b/pkg/connector/wadb/upgrades/00-latest-schema.sql index c4172d09..cd02333f 100644 --- a/pkg/connector/wadb/upgrades/00-latest-schema.sql +++ b/pkg/connector/wadb/upgrades/00-latest-schema.sql @@ -1,4 +1,4 @@ --- v0 -> v11 (compatible with v3+): Latest revision +-- v0 -> v12 (compatible with v3+): Latest revision CREATE TABLE whatsapp_poll_option_id ( bridge_id TEXT NOT NULL, @@ -50,6 +50,8 @@ CREATE TABLE whatsapp_history_sync_message ( REFERENCES whatsapp_history_sync_conversation (bridge_id, user_login_id, chat_jid) ON UPDATE CASCADE ON DELETE CASCADE ); +CREATE INDEX whatsapp_history_sync_message_chat_ts_idx ON whatsapp_history_sync_message (bridge_id, user_login_id, chat_jid, timestamp); + CREATE TABLE whatsapp_media_backfill_request ( bridge_id TEXT NOT NULL, user_login_id TEXT NOT NULL, diff --git a/pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql b/pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql new file mode 100644 index 00000000..0d3b910a --- /dev/null +++ b/pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql @@ -0,0 +1,2 @@ +-- v12 (compatible with v3+): Add index for history sync message timestamp lookups +CREATE INDEX whatsapp_history_sync_message_chat_ts_idx ON whatsapp_history_sync_message (bridge_id, user_login_id, chat_jid, timestamp); From bd0a26cedb6426daff8a66d2c7c82feded23264b Mon Sep 17 00:00:00 2001 From: "indent[bot]" <297468268+indent[bot]@users.noreply.github.com> Date: Sat, 14 Nov 2026 08:04:37 -0700 Subject: [PATCH 5/5] backfill: don't requeue history sync conversations that are still being created (#1035) Co-authored-by: Indent --- pkg/connector/backfill.go | 11 +++++++++++ pkg/connector/client.go | 2 ++ 2 files changed, 13 insertions(+) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index e2c7d114..e7748ac8 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -405,6 +405,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"). @@ -448,6 +450,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 queuedAt, ok := wa.historySyncQueued.Get(conv.ChatJID); ok && time.Since(queuedAt) < historySyncQueuedTimeout { + wg.Done() + continue } // TODO can the chat info fetch be avoided entirely? select { @@ -488,6 +493,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } + wa.historySyncQueued.Set(conv.ChatJID, time.Now()) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -503,6 +509,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.Delete(conv.ChatJID) wg.Done() }, }, @@ -510,6 +517,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { + wa.historySyncQueued.Delete(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -517,6 +525,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.Len() > 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 678c8d22..9b3f190a 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.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), @@ -114,6 +115,7 @@ type WhatsAppClient struct { MC mClient historySyncWakeup chan struct{} + historySyncQueued *exsync.Map[types.JID, time.Time] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex