backfill: log stub and message type of unparseable history messages (#1034)

Co-authored-by: Indent <noreply@indent.com>

.github/workflows/go.yml

@@ -30,13 +30,8 @@ jobs:
         run: |
           go install golang.org/x/tools/cmd/goimports@latest
           go install honnef.co/go/tools/cmd/staticcheck@latest
+          go install github.com/beeper/pre-commit-go/bridgelint@latest
           export PATH="$HOME/go/bin:$PATH"
 
-      - name: Cache pre-commit
-        uses: actions/cache@v6
-        with:
-          path: ~/.cache/pre-commit
-          key: pre-commit-go${{ matrix.go-version }}-${{ hashFiles('.pre-commit-config.yaml') }}
-
       - name: Run pre-commit
-        run: pipx run pre-commit run --all-files --show-diff-on-failure --color=always
+        uses: pre-commit/action@v3.0.1

.pre-commit-config.yaml

@@ -28,4 +28,5 @@ repos:
       - id: prevent-literal-http-methods
       - id: zerolog-ban-global-log
       - id: bridgelint
+        language: system
         verbose: true

go.mod

@@ -20,7 +20,7 @@ require (
 	golang.org/x/sync v0.24.0
 	google.golang.org/protobuf v1.36.12
 	gopkg.in/yaml.v3 v3.0.1
-	maunium.net/go/mautrix v0.32.1-0.20261031224934-b69deb427bd7
+	maunium.net/go/mautrix v0.32.1-0.20261105141340-842796acf445
 )
 
 require (

go.sum

@@ -103,5 +103,5 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 maunium.net/go/mauflag v1.0.0 h1:YiaRc0tEI3toYtJMRIfjP+jklH45uDHtT80nUamyD4M=
 maunium.net/go/mauflag v1.0.0/go.mod h1:nLivPOpTpHnpzEh8jEdSL9UqO9+/KBJFmNRlwKfkPeA=
-maunium.net/go/mautrix v0.32.1-0.20261031224934-b69deb427bd7 h1:GpLIqeJTpG0/ti7XA370HgjnSWd/pQxXVGb/FRkayOA=
-maunium.net/go/mautrix v0.32.1-0.20261031224934-b69deb427bd7/go.mod h1:tZSwPzw+/0pa0dxGufkh0uJqM84BNxB6e2I0+S0UyjU=
+maunium.net/go/mautrix v0.32.1-0.20261105141340-842796acf445 h1:ZI9Y2YEmjqQDloOCiciHeynTRYjUwK0ZZbNcRH/5NOc=
+maunium.net/go/mautrix v0.32.1-0.20261105141340-842796acf445/go.mod h1:tZSwPzw+/0pa0dxGufkh0uJqM84BNxB6e2I0+S0UyjU=

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()).

pkg/connector/handlematrix.go

@@ -319,7 +319,7 @@ func (wa *WhatsAppClient) HandleMatrixMessageRemove(ctx context.Context, msg *br
 }
 
 func (wa *WhatsAppClient) HandleMatrixReadReceipt(ctx context.Context, receipt *bridgev2.MatrixReadReceipt) (retErr error) {
-	if !receipt.ReadUpTo.After(receipt.LastRead) {
+	if receipt.Receipt.IsThread() || !receipt.ReadUpTo.After(receipt.LastRead) {
 		return nil
 	}
 	if receipt.LastRead.IsZero() {

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).

pkg/connector/lidmigrate.go

@@ -300,7 +300,11 @@ func (wa *WhatsAppConnector) syncMismatchingGhosts(ctx context.Context) error {
 }
 
 func (wa *WhatsAppConnector) removeRejoinedPNGhosts(ctx context.Context) error {
-	if wa.Bridge.Background || wa.Bridge.DB.KV.Get(ctx, "whatsapp_lid_dm_pn_ghosts_removed") == "true" {
+	if wa.Bridge.Background || wa.Bridge.DB.KV.Get(ctx, "whatsapp_lid_dm_pn_ghosts_removed_v2") == "true" {
+		return nil
+	}
+	stateGetter, ok := wa.Bridge.Matrix.(bridgev2.MatrixConnectorWithArbitraryRoomState)
+	if !ok {
 		return nil
 	}
 	log := zerolog.Ctx(ctx).With().Str("action", "remove rejoined pn ghosts").Logger()
@@ -322,12 +326,14 @@ func (wa *WhatsAppConnector) removeRejoinedPNGhosts(ctx context.Context) error {
 			continue
 		}
 		pnUserID := waid.MakeUserID(pn)
-		member, err := wa.Bridge.Matrix.GetMemberInfo(ctx, portal.MXID, wa.Bridge.Matrix.FormatGhostMXID(pnUserID))
-		if err != nil {
+		memberEvt, err := stateGetter.GetStateEvent(ctx, portal.MXID, event.StateMember, wa.Bridge.Matrix.FormatGhostMXID(pnUserID).String())
+		if errors.Is(err, mautrix.MNotFound) || errors.Is(err, mautrix.MForbidden) {
+			continue
+		} else if err != nil {
 			log.Warn().Err(err).Stringer("room_id", portal.MXID).Msg("Failed to get PN ghost membership")
 			failed++
 			continue
-		} else if member == nil || member.Membership != event.MembershipJoin {
+		} else if memberEvt == nil || memberEvt.Content.AsMember().Membership != event.MembershipJoin {
 			continue
 		}
 		pnGhost, err := wa.Bridge.GetGhostByID(ctx, pnUserID)
@@ -345,6 +351,12 @@ func (wa *WhatsAppConnector) removeRejoinedPNGhosts(ctx context.Context) error {
 			}
 			continue
 		}
+		log.Debug().
+			Stringer("room_id", portal.MXID).
+			Stringer("pn", pn).
+			Stringer("join_sender", memberEvt.Sender).
+			Time("join_ts", time.UnixMilli(memberEvt.Timestamp)).
+			Msg("Removed PN ghost from LID DM")
 		removed++
 	}
 	log.Info().
@@ -353,7 +365,7 @@ func (wa *WhatsAppConnector) removeRejoinedPNGhosts(ctx context.Context) error {
 		Int("total_lid_portals", len(lidPortals)).
 		Msg("Finished removing rejoined PN ghosts from LID DMs")
 	if failed == 0 && ctx.Err() == nil {
-		wa.Bridge.DB.KV.Set(ctx, "whatsapp_lid_dm_pn_ghosts_removed", "true")
+		wa.Bridge.DB.KV.Set(ctx, "whatsapp_lid_dm_pn_ghosts_removed_v2", "true")
 	}
 	return nil
 }

pkg/connector/muse.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"runtime/debug"
 	"time"
@@ -57,9 +56,9 @@ func (wa *WhatsAppClient) resyncWASARootSecrets(ctx context.Context) {
 	success := true
 	if err := wa.Client.FetchAppState(ctx, appstate.WAPatchRegularHigh, true, false); err != nil {
 		log.Err(err).Msg("Failed to resync WASA root secrets")
-		// Mark the sync as completed on mismatching LTHash errors, because those can get permanently stuck
-		// (retrying won't help). In case it's not stuck, a single call should be enough to request a resync.
-		if !errors.Is(err, appstate.ErrMismatchingLTHash) || ctx.Err() != nil {
+		// Mark the sync as completed on errors too, because some get permanently stuck (mismatching LTHash,
+		// expired patch blobs), and a failed full sync already requests app state recovery from the phone.
+		if ctx.Err() != nil {
 			return
 		}
 		success = false

pkg/connector/userinfo.go

@@ -119,6 +119,7 @@ func (wa *WhatsAppClient) doGhostResync(ctx context.Context, queue map[types.JID
 	var ghostJIDs []types.JID
 	var ghosts []*bridgev2.Ghost
 	var portals []*bridgev2.Portal
+	selectedGhosts := make(map[types.JID]*bridgev2.Ghost)
 	for jid, item := range queue {
 		var lastSync time.Time
 		unknownAvatar := false
@@ -136,12 +137,23 @@ func (wa *WhatsAppClient) doGhostResync(ctx context.Context, queue map[types.JID
 			continue
 		}
 		if item.ghost != nil {
-			ghosts = append(ghosts, item.ghost)
-			ghostJIDs = append(ghostJIDs, jid)
+			selectedGhosts[jid] = item.ghost
 		} else if item.portal != nil {
 			portals = append(portals, item.portal)
 		}
 	}
+	for jid, ghost := range selectedGhosts {
+		if jid.Server == types.DefaultUserServer {
+			lid, err := wa.GetStore().LIDs.GetLIDForPN(ctx, jid)
+			if err != nil {
+				log.Warn().Err(err).Stringer("jid", jid).Msg("Failed to get LID for background sync")
+			} else if _, lidSelected := selectedGhosts[lid]; lidSelected {
+				continue
+			}
+		}
+		ghosts = append(ghosts, ghost)
+		ghostJIDs = append(ghostJIDs, jid)
+	}
 	for _, portal := range portals {
 		wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{
 			EventMeta: simplevent.EventMeta{

pkg/msgconv/wa-rich-response.go

@@ -29,6 +29,7 @@ import (
 	"github.com/yuin/goldmark/extension"
 	"go.mau.fi/whatsmeow/types/richresponse"
 	"maunium.net/go/mautrix/bridgev2"
+	"maunium.net/go/mautrix/crypto/canonicaljson"
 	"maunium.net/go/mautrix/event"
 	"maunium.net/go/mautrix/format"
 	"maunium.net/go/mautrix/format/mdext"
@@ -36,7 +37,8 @@ import (
 
 func (mc *MessageConverter) convertUnifiedRichResponseMessage(ctx context.Context, data []byte) *bridgev2.ConvertedMessagePart {
 	extra := map[string]any{}
-	if json.Valid(data) && len(data) < 24*1024 {
+	_, canonicalErr := canonicaljson.Marshal(json.RawMessage(data))
+	if canonicalErr == nil && len(data) < 24*1024 {
 		extra["fi.mau.whatsapp.rich_message"] = json.RawMessage(data)
 	} else if len(data) < 16*1024 {
 		extra["fi.mau.whatsapp.non_json_rich_message"] = base64.StdEncoding.EncodeToString(data)