diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index 15662a93..e2c7d114 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -293,11 +293,17 @@ func (wa *WhatsAppClient) handleWAHistorySync( for i, rawMsg := range rawMessages { // Don't store messages that will just be skipped. msgEvt, err := wa.Client.ParseWebMessage(jid, rawMsg.GetMessage()) - if err != nil { + if err != nil && getMessageType(rawMsg.GetMessage().GetMessage()) == "ignore" { + ignoredTypes++ + continue + } else if err != nil { log.Warn().Err(err). 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()). @@ -566,15 +572,6 @@ 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 @@ -583,11 +580,7 @@ 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) { - if startTime != nil { - wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, uint64(startTime.Unix())) - } else { - wa.deleteHistorySyncMessages(ctx, portalJID, lastMessageTS, 0) - } + wa.deleteHistorySyncMessages(ctx, portalJID, 0, 0) hasMore = hasMore && params.AnchorMessage != nil if hasMore && !params.AllowSlowFetch { return &bridgev2.FetchMessagesResponse{ @@ -615,14 +608,7 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet } } } - var deleteUpTo uint64 - if !params.Forward && !hasMore { - deleteUpTo = 1 << 62 - if lastMessageTS != 0 { - deleteUpTo = lastMessageTS - } - } - resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, deleteUpTo) + resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, !params.Forward && !hasMore) if err != nil { return nil, fmt.Errorf("failed to convert messages: %w", err) } @@ -684,7 +670,7 @@ func (wa *WhatsAppClient) convertHistorySyncMessages( portalJID types.JID, messages []*waWeb.WebMessageInfo, explodeOnError bool, - deleteUpTo uint64, + deleteEverything bool, ) (*bridgev2.FetchMessagesResponse, error) { cli := wa.Client if cli == nil { @@ -743,7 +729,11 @@ func (wa *WhatsAppClient) convertHistorySyncMessages( Messages: convertedMessages, Cursor: networkid.PaginationCursor(strconv.FormatUint(oldestTS, 10)), CompleteCallback: func() { - wa.deleteHistorySyncMessages(ctx, portalJID, max(newestTS, deleteUpTo), oldestTS) + if deleteEverything { + wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, oldestTS) + } else { + wa.deleteHistorySyncMessages(ctx, portalJID, newestTS, oldestTS) + } if len(mediaRequests) > 0 { go func(ctx context.Context) { for _, req := range mediaRequests { @@ -906,7 +896,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, 0) + resp, err := wa.convertHistorySyncMessages(ctx, portal, portalJID, messages, false, false) if err != nil { return nil, err } diff --git a/pkg/connector/handlewhatsapp.go b/pkg/connector/handlewhatsapp.go index 4391233b..e6174b92 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). diff --git a/pkg/msgconv/from-whatsapp.go b/pkg/msgconv/from-whatsapp.go index db18f35a..ec273076 100644 --- a/pkg/msgconv/from-whatsapp.go +++ b/pkg/msgconv/from-whatsapp.go @@ -322,6 +322,7 @@ func (mc *MessageConverter) ToMatrix( cm.ThreadRoot = ptr.Ptr(waid.MakeMessageID(chat, pcp, commentTarget.GetID())) } mc.cutOffLongMessage(part.Content) + logOversizedPart(ctx, part) return cm } diff --git a/pkg/msgconv/wa-text.go b/pkg/msgconv/wa-text.go index 52874fd8..7afcc81f 100644 --- a/pkg/msgconv/wa-text.go +++ b/pkg/msgconv/wa-text.go @@ -18,6 +18,7 @@ package msgconv import ( "context" + "encoding/json" "fmt" "html" "regexp" @@ -25,6 +26,7 @@ import ( "strconv" "strings" + "github.com/rs/zerolog" "go.mau.fi/whatsmeow/proto/waE2E" "maunium.net/go/mautrix/bridgev2" "maunium.net/go/mautrix/event" @@ -72,6 +74,36 @@ func (mc *MessageConverter) cutOffLongMessage(content *event.MessageEventContent } } +func logOversizedPart(ctx context.Context, part *bridgev2.ConvertedMessagePart) { + if part == nil || part.Content == nil { + return + } + contentJSON, _ := json.Marshal(part.Content) + extraJSON, _ := json.Marshal(part.Extra) + if len(contentJSON)+len(extraJSON) <= 40000 { + return + } + extraKeys := make([]string, 0, len(part.Extra)) + for key := range part.Extra { + extraKeys = append(extraKeys, key) + } + var mentionCount int + if part.Content.Mentions != nil { + mentionCount = len(part.Content.Mentions.UserIDs) + } + linkPreviewJSON, _ := json.Marshal(part.Content.BeeperLinkPreviews) + zerolog.Ctx(ctx).Warn(). + Str("msg_type", string(part.Content.MsgType)). + Int("content_json_size", len(contentJSON)). + Int("extra_json_size", len(extraJSON)). + Int("body_size", len(part.Content.Body)). + Int("formatted_body_size", len(part.Content.FormattedBody)). + Int("mention_count", mentionCount). + Int("link_preview_json_size", len(linkPreviewJSON)). + Strs("extra_keys", extraKeys). + Msg("Converted message part is larger than the cut-off limit") +} + func (mc *MessageConverter) parseFormatting(content *event.MessageEventContent, allowInlineURL, forceHTML bool) { parsedHTML := parseWAFormattingToHTML(content.Body, allowInlineURL) if forceHTML || parsedHTML != event.TextToHTML(content.Body) {