diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index e2a18b18..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()). @@ -399,8 +405,6 @@ 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"). @@ -444,9 +448,6 @@ 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 { @@ -487,7 +488,6 @@ 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,7 +503,6 @@ 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() }, }, @@ -511,7 +510,6 @@ 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 } @@ -519,9 +517,6 @@ 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 9b3f190a..678c8d22 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -50,7 +50,6 @@ 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), @@ -115,7 +114,6 @@ 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 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) {