xchat: fixed nullable conversation IDs blocking catch-up

pkg/connector/xchat_reconnect.go

@@ -189,8 +189,13 @@ func (tc *TwitterClient) processXChatInboxPage(
 	}
 	log := zerolog.Ctx(ctx)
 	var pageMissing []string
+	unavailableCount := 0
 	for i := range page.Items {
 		item := &page.Items[i]
+		if item.ConversationUnavailable {
+			unavailableCount++
+			continue
+		}
 		if item.ConversationDetail.ConversationID == "" {
 			return nil, fmt.Errorf("XChat inbox item %d has no conversation ID", i)
 		}
@@ -199,6 +204,9 @@ func (tc *TwitterClient) processXChatInboxPage(
 		}
 		pageMissing = append(pageMissing, tc.cacheUsersFromItem(item)...)
 	}
+	if unavailableCount > 0 {
+		log.Warn().Int("unavailable_conversations", unavailableCount).Msg("Skipping unavailable XChat conversations")
+	}
 
 	if len(pageMissing) > 0 {
 		if err := tc.ensureUsersInCacheByID(ctx, pageMissing); err != nil {
@@ -214,6 +222,9 @@ func (tc *TwitterClient) processXChatInboxPage(
 	g.SetLimit(10)
 	for i := range page.Items {
 		item := &page.Items[i]
+		if item.ConversationUnavailable {
+			continue
+		}
 		g.Go(func() error {
 			conversationID := item.ConversationDetail.ConversationID
 			keyErr := processor.ProcessKeyChangeEvents(pageCtx, item)

pkg/twittermeow/data/response/xchat.go

@@ -1,5 +1,11 @@
 package response
 
+import (
+	"encoding/json"
+
+	"github.com/tidwall/gjson"
+)
+
 type XChatGraphQLError struct {
 	Message string `json:"message,omitempty"`
 }
@@ -107,6 +113,7 @@ type XChatInboxCursor struct {
 }
 
 type XChatInboxItem struct {
+	ConversationUnavailable            bool                        `json:"-"`
 	Typename                           string                      `json:"__typename,omitempty"`
 	LatestMessageEvents                []string                    `json:"latest_message_events,omitempty"`
 	EncodedMessageEvents               []string                    `json:"encoded_message_events,omitempty"`
@@ -117,6 +124,43 @@ type XChatInboxItem struct {
 	HasMore                            bool                        `json:"has_more,omitempty"`
 }
 
+func (item *XChatInboxItem) UnmarshalJSON(data []byte) error {
+	type plain XChatInboxItem
+	var decoded plain
+	if err := json.Unmarshal(data, &decoded); err != nil {
+		return err
+	}
+	*item = XChatInboxItem(decoded)
+	if item.ConversationDetail.ConversationID != "" {
+		return nil
+	}
+	raw := gjson.ParseBytes(data)
+	events := raw.Get("latest_message_events")
+	detail := raw.Get("conversation_detail")
+	if raw.Get("__typename").Str != "XChatGetInboxPageConversationData" || !events.IsArray() || !detail.IsObject() {
+		return nil
+	}
+	switch detail.Get("__typename").Str {
+	case "XChatGroupConversationDetail", "XChatDirectConversationDetail":
+	default:
+		return nil
+	}
+	conversationID := detail.Get("conversation_id")
+	if !conversationID.Exists() || (conversationID.Type != gjson.Null &&
+		(conversationID.Type != gjson.String || conversationID.Str != "")) {
+		return nil
+	}
+	for _, rawEvent := range events.Array() {
+		if rawEvent.Type != gjson.String {
+			return nil
+		}
+	}
+	// X's client omits known conversation records whose nullable ID is unavailable.
+	// Keep malformed required fields distinct so they still block the checkpoint.
+	item.ConversationUnavailable = true
+	return nil
+}
+
 type XChatConversationDetail struct {
 	Typename                           string              `json:"__typename,omitempty"`
 	IsMuted                            bool                `json:"is_muted,omitempty"`

pkg/twittermeow/messaging.go

@@ -718,6 +718,9 @@ func (c *Client) getXChatInboxPage(ctx context.Context, url string, variables Fo
 	if !isSnapshot && !isDelta {
 		return nil, fmt.Errorf("%s returned an unsupported inbox page shape", opName)
 	}
+	if isSnapshot {
+		c.logXChatInboxItemShape(pageBody, &page, "data."+responseKey)
+	}
 	cursorPresent, pullFinished := page.InboxCursor.CursorID != "", page.InboxCursor.PullFinished
 	if cursor := page.MessageEventsCursor; cursor != nil {
 		cursorPresent, pullFinished = cursor.MaxLocalSequenceID != "", cursor.PullFinished

pkg/twittermeow/messaging_inbox_diagnostics.go

@@ -0,0 +1,58 @@
+package twittermeow
+
+import (
+	"strconv"
+
+	"github.com/tidwall/gjson"
+
+	"go.mau.fi/mautrix-twitter/pkg/twittermeow/data/response"
+)
+
+func (c *Client) logXChatInboxItemShape(rawPage gjson.Result, page *response.XChatInboxPage, decoderPath string) {
+	for i, item := range page.Items {
+		if item.ConversationUnavailable || item.ConversationDetail.ConversationID != "" {
+			continue
+		}
+		// Typed decoding collapses absent/null objects and empty IDs. Inspect only
+		// the first item that the connector will reject, without retaining values.
+		rawItem := rawPage.Get("items." + strconv.Itoa(i))
+		detail := rawItem.Get("conversation_detail")
+		c.Logger.Warn().
+			Str("decoder_path", decoderPath).
+			Str("page_kind", inboxJSONKind(rawPage)).
+			Int("item_index", i).Int("item_count", len(page.Items)).
+			Str("item_kind", inboxJSONKind(rawItem)).
+			Int("item_field_count", len(rawItem.Map())).
+			Bool("item_typename_present", rawItem.Get("__typename").Exists()).
+			Str("detail_kind", inboxJSONKind(detail)).
+			Int("detail_field_count", len(detail.Map())).
+			Bool("detail_typename_present", detail.Get("__typename").Exists()).
+			Str("conversation_id_kind", inboxJSONKind(detail.Get("conversation_id"))).
+			Int("page_error_count", len(page.Errors)).
+			Int("latest_message_count", len(item.LatestMessageEvents)).
+			Int("encoded_message_count", len(item.EncodedMessageEvents)).
+			Bool("notifiable_message_present", item.LatestNotifiableMessageCreateEvent != "").
+			Int("key_change_count", len(item.LatestConversationKeyChangeEvents)).
+			Int("read_event_count", len(item.LatestReadEventsPerParticipant)).
+			Bool("has_more", item.HasMore).
+			Msg("XChat inbox item has no conversation ID")
+		return
+	}
+}
+
+func inboxJSONKind(value gjson.Result) string {
+	switch {
+	case !value.Exists():
+		return "missing"
+	case value.Type == gjson.Null:
+		return "null"
+	case value.IsObject():
+		return "object"
+	case value.IsArray():
+		return "array"
+	case value.Type == gjson.String && value.Str == "":
+		return "empty_string"
+	default:
+		return value.Type.String()
+	}
+}