chatsync: classify requests using trust state

pkg/connector/chatsync.go

@@ -51,6 +51,96 @@ func isXChatPortalForLogin(portal *bridgev2.Portal, loginID networkid.UserLoginI
 	return ok && meta.IsXChatConversation()
 }
 
+func xchatInboxItemTrust(item *response.XChatInboxItem) *bool {
+	defer func() { _ = recover() }()
+	if item == nil {
+		return nil
+	}
+	encodedEvents := make([]string, 0, len(item.LatestMessageEvents)+len(item.EncodedMessageEvents)+len(item.LatestConversationKeyChangeEvents)+1)
+	encodedEvents = append(encodedEvents, item.LatestMessageEvents...)
+	encodedEvents = append(encodedEvents, item.EncodedMessageEvents...)
+	encodedEvents = append(encodedEvents, item.LatestConversationKeyChangeEvents...)
+	if item.LatestNotifiableMessageCreateEvent != "" {
+		encodedEvents = append(encodedEvents, item.LatestNotifiableMessageCreateEvent)
+	}
+
+	var trusted *bool
+	latestSequenceID := ""
+	for _, encoded := range encodedEvents {
+		decoded, err := base64.StdEncoding.DecodeString(encoded)
+		if err != nil {
+			continue
+		}
+		var evt payload.MessageEvent
+		if err = payload.Decode(decoded, &evt); err != nil || evt.IsTrusted == nil {
+			continue
+		}
+		sequenceID := ptr.Val(evt.SequenceId)
+		if trusted != nil && (sequenceID == "" && latestSequenceID != "" ||
+			sequenceID != "" && latestSequenceID != "" && compareIntStrings(sequenceID, latestSequenceID) < 0) {
+			continue
+		}
+		value := *evt.IsTrusted
+		trusted = &value
+		latestSequenceID = sequenceID
+	}
+	return trusted
+}
+
+func applyXChatTrustToChatInfo(info *bridgev2.ChatInfo, trusted bool) {
+	info.MessageRequest = ptr.Ptr(!trusted)
+	info.ExtraUpdates = bridgev2.MergeExtraUpdaters(info.ExtraUpdates, func(_ context.Context, portal *bridgev2.Portal) bool {
+		meta, ok := portal.Metadata.(*PortalMetadata)
+		if !ok || meta == nil || (meta.XChatTrusted != nil && *meta.XChatTrusted == trusted) {
+			return false
+		}
+		meta.XChatTrusted = ptr.Ptr(trusted)
+		return true
+	})
+}
+
+func (tc *TwitterClient) syncXChatTrust(
+	ctx context.Context,
+	conversationID string,
+	trusted *bool,
+	timestamp time.Time,
+	streamOrder int64,
+) bool {
+	if trusted == nil {
+		return true
+	}
+	if ctx == nil {
+		ctx = context.TODO()
+	}
+	portalKey := tc.MakePortalKeyFromID(conversationID)
+	portal, err := tc.connector.br.GetPortalByKey(ctx, portalKey)
+	if err != nil {
+		zerolog.Ctx(ctx).Warn().Err(err).Msg("Failed to get XChat portal for trust update")
+		return false
+	}
+	meta, _ := portal.Metadata.(*PortalMetadata)
+	messageRequest := !*trusted
+	if meta != nil && meta.XChatTrusted != nil && *meta.XChatTrusted == *trusted && portal.MessageRequest == messageRequest {
+		return true
+	}
+
+	chatInfo := &bridgev2.ChatInfo{}
+	applyXChatTrustToChatInfo(chatInfo, *trusted)
+	if current := bridgev2.GetRemoteEventFromContext(ctx); current != nil && current.GetPortalKey().ID == portalKey.ID {
+		portal.UpdateInfo(ctx, chatInfo, tc.userLogin, nil, timestamp)
+		return true
+	}
+	return tc.queueXChatRemoteEventWithPortalRepair(ctx, conversationID, &simplevent.ChatInfoChange{
+		EventMeta: simplevent.EventMeta{
+			Type:        bridgev2.RemoteEventChatInfoChange,
+			PortalKey:   portalKey,
+			Timestamp:   timestamp,
+			StreamOrder: streamOrder,
+		},
+		ChatInfoChange: &bridgev2.ChatInfoChange{ChatInfo: chatInfo},
+	})
+}
+
 // TODO: Remove this repair after affected bridges have been reconnected
 func (tc *TwitterClient) repairExistingXChatMessageRequests(ctx context.Context) error {
 	if tc.xchatRequestsRepaired {
@@ -68,8 +158,13 @@ func (tc *TwitterClient) repairExistingXChatMessageRequests(ctx context.Context)
 		}
 		xchatRooms++
 
+		meta := portal.Metadata.(*PortalMetadata)
+		messageRequest := false
+		if meta.XChatTrusted != nil {
+			messageRequest = !*meta.XChatTrusted
+		}
 		wasMessageRequest := portal.MessageRequest
-		portal.MessageRequest = false
+		portal.MessageRequest = messageRequest
 		internals := (*bridgev2.PortalInternals)(portal)
 		stateKey, bridgeInfo := internals.GetBridgeInfo()
 		for _, eventType := range []event.Type{event.StateBridge, event.StateHalfShotBridge} {
@@ -81,7 +176,7 @@ func (tc *TwitterClient) repairExistingXChatMessageRequests(ctx context.Context)
 				return fmt.Errorf("repair XChat message-request room state")
 			}
 		}
-		if !wasMessageRequest {
+		if wasMessageRequest == messageRequest {
 			continue
 		}
 		if err = portal.Save(ctx); err != nil {
@@ -102,6 +197,9 @@ func shouldEmitChatInfoUpdate(chatInfo *bridgev2.ChatInfo, portalRoomType databa
 	if chatInfo == nil {
 		return false
 	}
+	if chatInfo.MessageRequest != nil || chatInfo.ExtraUpdates != nil {
+		return true
+	}
 
 	// DM room title/avatar are member-derived, so always emit ChatInfoChange
 	// for existing DM rooms to refresh stale member profile info.
@@ -242,9 +340,12 @@ func (tc *TwitterClient) xchatItemToConversation(ctx context.Context, item *resp
 
 	conv := &types.Conversation{
 		ConversationID: detail.ConversationID,
-		Trusted:        true, // XChat conversations are always trusted
+		Trusted:        true,
 		Muted:          detail.IsMuted,
 	}
+	if trusted := xchatInboxItemTrust(item); trusted != nil {
+		conv.Trusted = *trusted
+	}
 
 	// Determine conversation type based on conversation ID and metadata.
 	if strings.HasPrefix(detail.ConversationID, "g") || detail.GroupMetadata != nil ||
@@ -384,11 +485,7 @@ func (tc *TwitterClient) xchatItemToChatInfo(ctx context.Context, item *response
 		}
 	}
 
-	// MessageRequest is true for untrusted conversations (message requests)
-	var messageRequest *bool
-	if conv != nil {
-		messageRequest = ptr.Ptr(!conv.Trusted)
-	}
+	trusted := xchatInboxItemTrust(item)
 	membersAreFull := len(detail.GroupMembersResults) > 0 && len(memberMap) > 0
 	for _, memberResult := range detail.GroupMembersResults {
 		if memberID, _ := xchatUserFromResult(memberResult); memberID == "" {
@@ -418,8 +515,10 @@ func (tc *TwitterClient) xchatItemToChatInfo(ctx context.Context, item *response
 			TotalMemberCount: len(memberMap),
 			MemberMap:        memberMap,
 		},
-		CanBackfill:    true,
-		MessageRequest: messageRequest,
+		CanBackfill: true,
+	}
+	if trusted != nil {
+		applyXChatTrustToChatInfo(info, *trusted)
 	}
 
 	if isGroup {

pkg/connector/chatsync_test.go

@@ -2,11 +2,53 @@ package connector
 
 import (
 	"context"
+	"encoding/base64"
 	"testing"
 
+	"go.mau.fi/util/ptr"
+	"maunium.net/go/mautrix/bridgev2"
+	"maunium.net/go/mautrix/bridgev2/database"
+
+	"go.mau.fi/mautrix-twitter/pkg/twittermeow/data/payload"
 	"go.mau.fi/mautrix-twitter/pkg/twittermeow/data/response"
 )
 
+func TestXChatItemTrustControlsMessageRequest(t *testing.T) {
+	for _, trusted := range []*bool{ptr.Ptr(false), ptr.Ptr(true), nil} {
+		client := &TwitterClient{}
+		item := &response.XChatInboxItem{
+			ConversationDetail:  response.XChatConversationDetail{ConversationID: "g123"},
+			LatestMessageEvents: []string{"AgABMA=="},
+		}
+		if trusted != nil {
+			encoded, err := payload.Encode(&payload.MessageEvent{IsTrusted: trusted})
+			if err != nil {
+				t.Fatal(err)
+			}
+			item.LatestMessageEvents = []string{base64.StdEncoding.EncodeToString(encoded)}
+		}
+		conv := client.xchatItemToConversation(t.Context(), item, nil)
+		if conv.Trusted != (trusted == nil || *trusted) {
+			t.Fatalf("Trusted = %t", conv.Trusted)
+		}
+		info := client.xchatItemToChatInfo(t.Context(), item, nil, conv)
+		if trusted == nil {
+			if info.MessageRequest != nil || info.ExtraUpdates != nil {
+				t.Fatal("missing trust changed message-request state")
+			}
+			return
+		}
+		if info.MessageRequest == nil || *info.MessageRequest == *trusted {
+			t.Fatalf("MessageRequest = %v", info.MessageRequest)
+		}
+		meta := &PortalMetadata{}
+		portal := &bridgev2.Portal{Portal: &database.Portal{Metadata: meta}}
+		if !info.ExtraUpdates(t.Context(), portal) || meta.XChatTrusted == nil || *meta.XChatTrusted != *trusted {
+			t.Fatalf("XChatTrusted = %v", meta.XChatTrusted)
+		}
+	}
+}
+
 func TestXChatItemToConversationPreservesPlaintextGroupName(t *testing.T) {
 	tc := &TwitterClient{}
 	item := &response.XChatInboxItem{

pkg/connector/dbmeta.go

@@ -51,6 +51,7 @@ type PortalMetadata struct {
 
 	// Server token for XChat API
 	ConversationToken string `json:"conversation_token,omitempty"`
+	XChatTrusted      *bool  `json:"xchat_trusted,omitempty"`
 }
 
 // CanUseXChat returns true if this conversation has encryption keys

pkg/connector/handlematrix.go

@@ -1022,5 +1022,13 @@ func (tc *TwitterClient) HandleMatrixAcceptMessageRequest(ctx context.Context, m
 	if errors.Is(err, twittermeow.ErrConversationDoesntExist) {
 		err = nil
 	}
+	if err == nil && msg != nil && msg.Portal != nil {
+		meta, ok := msg.Portal.Metadata.(*PortalMetadata)
+		if ok && meta.IsXChatConversation() {
+			chatInfo := &bridgev2.ChatInfo{}
+			applyXChatTrustToChatInfo(chatInfo, true)
+			msg.Portal.UpdateInfo(ctx, chatInfo, tc.userLogin, nil, time.Time{})
+		}
+	}
 	return err
 }

pkg/connector/handletwit.go

@@ -229,6 +229,9 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit
 				return false
 			}
 		}
+		if !tc.syncXChatTrust(ctx, evt.ConversationID, evt.Trusted, methods.ParseMsecTimestamp(evt.Time), streamOrder) {
+			return false
+		}
 
 		txnID := evt.RequestID
 		if txnID == "" {
@@ -283,6 +286,9 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit
 				return false
 			}
 		}
+		if !tc.syncXChatTrust(ctx, evt.ConversationID, evt.Trusted, methods.ParseMsecTimestamp(evt.Time), streamOrder) {
+			return false
+		}
 
 		txnID := evt.RequestID
 		if txnID == "" {
@@ -536,6 +542,15 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit
 				Msg("Failed to ensure portal for ConversationCreate event")
 			return false
 		}
+		if !tc.syncXChatTrust(
+			ctx,
+			evt.ConversationID,
+			evt.Trusted,
+			methods.ParseMsecTimestamp(evt.Time),
+			methods.ParseSnowflakeInt(evt.ID),
+		) {
+			return false
+		}
 
 		// If the portal was just created or doesn't have a room yet, trigger a resync
 		// to ensure it gets created and backfilled
@@ -579,6 +594,7 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit
 				Msg("Failed to get chat info for trusted conversation")
 			return false
 		}
+		applyXChatTrustToChatInfo(chatInfo, true)
 
 		return xchatRemoteEventHandled(tc.userLogin.QueueRemoteEvent(&simplevent.ChatResync{
 			EventMeta: simplevent.EventMeta{

pkg/twittermeow/data/types/messaging.go

@@ -30,6 +30,7 @@ type Message struct {
 	SequenceID             string            `json:"sequence_id,omitempty"`
 	RequestID              string            `json:"request_id,omitempty"`
 	ConversationID         string            `json:"conversation_id,omitempty"`
+	Trusted                *bool             `json:"trusted,omitempty"`
 	ConversationKeyVersion string            `json:"conversation_key_version,omitempty"`
 	MessageData            MessageData       `json:"message_data,omitempty"`
 	MessageReactions       []MessageReaction `json:"message_reactions,omitempty"`
@@ -92,6 +93,7 @@ type ConversationCreate struct {
 	AffectsSort    bool   `json:"affects_sort,omitempty"`
 	ConversationID string `json:"conversation_id,omitempty"`
 	RequestID      string `json:"request_id,omitempty"`
+	Trusted        *bool  `json:"trusted,omitempty"`
 }
 
 type ConversationDelete struct {

pkg/twittermeow/xchat_convert.go

@@ -58,6 +58,7 @@ func convertXChatMessageToTwitterMessage(evt *payload.MessageEvent, contents *pa
 		SequenceID:             seqID,
 		RequestID:              ptr.Val(evt.MessageId),
 		ConversationID:         ptr.Val(evt.ConversationId),
+		Trusted:                evt.IsTrusted,
 		ConversationKeyVersion: keyVersion,
 		MessageData:            msgData,
 	}
@@ -96,6 +97,7 @@ func convertXChatMessageEdit(evt *payload.MessageEvent, edit *payload.MessageEdi
 		SequenceID:             ptr.Val(evt.SequenceId),
 		RequestID:              ptr.Val(evt.MessageId),
 		ConversationID:         ptr.Val(evt.ConversationId),
+		Trusted:                evt.IsTrusted,
 		ConversationKeyVersion: keyVersion,
 		MessageData:            msgData,
 	})

pkg/twittermeow/xchat_processor.go

@@ -546,6 +546,7 @@ func (p *XChatEventProcessor) emitConversationCreate(ctx context.Context, evt *p
 		Time:           ptr.Val(evt.CreatedAtMsec),
 		ConversationID: ptr.Val(evt.ConversationId),
 		RequestID:      ptr.Val(evt.MessageId),
+		Trusted:        evt.IsTrusted,
 	})
 }
 

pkg/twittermeow/xchat_send_test.go

@@ -11,9 +11,17 @@ import (
 	"github.com/rs/zerolog"
 
 	"go.mau.fi/mautrix-twitter/pkg/twittermeow/cookies"
+	"go.mau.fi/mautrix-twitter/pkg/twittermeow/data/payload"
 	"go.mau.fi/mautrix-twitter/pkg/twittermeow/data/response"
 )
 
+func TestConvertXChatMessagePreservesTrust(t *testing.T) {
+	trusted := false
+	if got := convertXChatMessageToTwitterMessage(&payload.MessageEvent{IsTrusted: &trusted}, &payload.MessageContents{}, ""); got.Trusted == nil || *got.Trusted {
+		t.Fatalf("Trusted = %v", got.Trusted)
+	}
+}
+
 func TestRefreshConversationKeysDoesNotBlockOnConversationDataCallback(t *testing.T) {
 	client := NewClient(cookies.NewCookies(nil), nil, zerolog.Nop())
 	client.HTTP = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {