From 3f5cd3a82e822727ae6a60ef5445ea9f1ab7d550 Mon Sep 17 00:00:00 2001 From: Gerardo Rodriguez Date: Sun, 8 Nov 2026 15:19:06 +0000 Subject: [PATCH 1/2] backfill: only delete history sync messages on the side that was fetched --- pkg/connector/backfill.go | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index cbc7eec..360d92a 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -574,7 +574,13 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet if err != nil { return nil, fmt.Errorf("failed to load messages from database: %w", err) } else if len(messages) == 0 || (len(messages) == 1 && anchorID != "" && messages[0].GetKey().GetID() == anchorID) { - wa.deleteHistorySyncMessages(ctx, portalJID, 0, 0) + if startTime != nil { + wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, uint64(startTime.Unix())) + } else if endTime != nil { + wa.deleteHistorySyncMessages(ctx, portalJID, uint64(endTime.Unix()), 0) + } else { + wa.deleteHistorySyncMessages(ctx, portalJID, 0, 0) + } hasMore = hasMore && params.AnchorMessage != nil if hasMore && !params.AllowSlowFetch { return &bridgev2.FetchMessagesResponse{ @@ -602,7 +608,14 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet } } } - resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, !params.Forward && !hasMore) + var deleteUpTo uint64 + if !params.Forward && !hasMore { + deleteUpTo = 1 << 62 + if endTime != nil { + deleteUpTo = uint64(endTime.Unix()) + } + } + resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, deleteUpTo) if err != nil { return nil, fmt.Errorf("failed to convert messages: %w", err) } @@ -664,7 +677,7 @@ func (wa *WhatsAppClient) convertHistorySyncMessages( portalJID types.JID, messages []*waWeb.WebMessageInfo, explodeOnError bool, - deleteEverything bool, + deleteUpTo uint64, ) (*bridgev2.FetchMessagesResponse, error) { cli := wa.Client if cli == nil { @@ -723,11 +736,7 @@ func (wa *WhatsAppClient) convertHistorySyncMessages( Messages: convertedMessages, Cursor: networkid.PaginationCursor(strconv.FormatUint(oldestTS, 10)), CompleteCallback: func() { - if deleteEverything { - wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, oldestTS) - } else { - wa.deleteHistorySyncMessages(ctx, portalJID, newestTS, oldestTS) - } + wa.deleteHistorySyncMessages(ctx, portalJID, max(newestTS, deleteUpTo), oldestTS) if len(mediaRequests) > 0 { go func(ctx context.Context) { for _, req := range mediaRequests { @@ -890,7 +899,7 @@ func (wa *WhatsAppClient) handleOnDemandHistorySync(ctx context.Context, blob *w Int("message_count", len(messages)). Stringer("end_of_history_type", conv.GetEndOfHistoryTransferType()). Msg("Converting messages to bridge from on-demand history sync") - resp, err := wa.convertHistorySyncMessages(ctx, portal, portalJID, messages, false, false) + resp, err := wa.convertHistorySyncMessages(ctx, portal, portalJID, messages, false, 0) if err != nil { return nil, err } From 2cd9ae28972060bfc2f0d12294ef3edc67ff7c0d Mon Sep 17 00:00:00 2001 From: Gerardo Rodriguez Date: Sun, 8 Nov 2026 15:38:45 +0000 Subject: [PATCH 2/2] backfill: delete history sync messages up to the newest bridged message on backward fetches --- pkg/connector/backfill.go | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index 360d92a..15662a9 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -566,6 +566,15 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet anchorID = parsedID.ID } } + var lastMessageTS uint64 + if !params.Forward { + lastMessage, err := wa.Main.Bridge.DB.Message.GetLastPartAtOrBeforeTime(ctx, params.Portal.PortalKey, time.Now()) + if err != nil { + return nil, fmt.Errorf("failed to get last message in portal: %w", err) + } else if lastMessage != nil { + lastMessageTS = uint64(lastMessage.Timestamp.Unix()) + } + } var hasMore bool if !params.Forward && wa.Main.Config.HistorySync.BackwardsOnDemand { hasMore = conv != nil && ptr.Val(conv.EndOfHistoryTransferType) == waHistorySync.Conversation_COMPLETE_BUT_MORE_MESSAGES_REMAIN_ON_PRIMARY @@ -576,10 +585,8 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet } else if len(messages) == 0 || (len(messages) == 1 && anchorID != "" && messages[0].GetKey().GetID() == anchorID) { if startTime != nil { wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, uint64(startTime.Unix())) - } else if endTime != nil { - wa.deleteHistorySyncMessages(ctx, portalJID, uint64(endTime.Unix()), 0) } else { - wa.deleteHistorySyncMessages(ctx, portalJID, 0, 0) + wa.deleteHistorySyncMessages(ctx, portalJID, lastMessageTS, 0) } hasMore = hasMore && params.AnchorMessage != nil if hasMore && !params.AllowSlowFetch { @@ -611,8 +618,8 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet var deleteUpTo uint64 if !params.Forward && !hasMore { deleteUpTo = 1 << 62 - if endTime != nil { - deleteUpTo = uint64(endTime.Unix()) + if lastMessageTS != 0 { + deleteUpTo = lastMessageTS } } resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, deleteUpTo)