diff --git a/pkg/connector/chatsync.go b/pkg/connector/chatsync.go index d386462..e578677 100644 --- a/pkg/connector/chatsync.go +++ b/pkg/connector/chatsync.go @@ -20,13 +20,11 @@ import ( "context" "encoding/base64" "strings" - "time" "github.com/rs/zerolog" "go.mau.fi/util/ptr" "maunium.net/go/mautrix/bridgev2" "maunium.net/go/mautrix/bridgev2/database" - "maunium.net/go/mautrix/bridgev2/simplevent" "go.mau.fi/mautrix-twitter/pkg/twittermeow" "go.mau.fi/mautrix-twitter/pkg/twittermeow/crypto" @@ -61,7 +59,7 @@ func shouldEmitChatInfoUpdate(chatInfo *bridgev2.ChatInfo, portalRoomType databa } // syncXChatChannel syncs a single conversation from XChat inbox data. -// Creates the portal synchronously if it doesn't exist. +// It queues a portal resync and lets bridgev2 create the room if needed. func (tc *TwitterClient) syncXChatChannel(ctx context.Context, item *response.XChatInboxItem, users map[string]*types.User) { log := zerolog.Ctx(ctx) @@ -94,54 +92,23 @@ func (tc *TwitterClient) syncXChatChannel(ctx context.Context, item *response.XC } } - // Ensure a backfill task exists even if we don't end up emitting a ChatInfoChange. - // Beeper scrollback relies on the backfill task existing for the portal. - if portal.MXID != "" { - if chatInfo.CanBackfill { - // FIXME this is wrong, backfill tasks are created automatically based on chat resyncs - if err := tc.connector.br.DB.BackfillTask.EnsureExists(ctx, portal.PortalKey, tc.userLogin.ID); err != nil { - log.Warn().Err(err). - Str("conversation_id", conv.ConversationID). - Msg("Failed to ensure backfill task exists") - } else { - tc.connector.br.WakeupBackfillQueue() - } - } - } - - // Create Matrix room if it doesn't exist + // Queue a ChatResync so bridgev2 owns room creation and backfill task registration. if portal.MXID == "" { - // FIXME this is wrong, CreateMatrixRoom should not be called manually - err = portal.CreateMatrixRoom(ctx, tc.userLogin, chatInfo) - if err != nil { - log.Warn().Err(err). + resync := tc.queueChatResyncNow(chatResyncCreate, portal.PortalKey, chatInfo) + if !resync.Success { + log.Warn(). Str("conversation_id", conv.ConversationID). - Msg("Failed to create Matrix room") + Err(resync.Error). + Msg("Failed to queue ChatResync for XChat conversation") return } - // Register backfill task for the newly created room - if chatInfo.CanBackfill { - // FIXME this is wrong, backfill tasks are created automatically based on chat resyncs - if err := tc.connector.br.DB.BackfillTask.EnsureExists(ctx, portal.PortalKey, tc.userLogin.ID); err != nil { - log.Warn().Err(err). - Str("conversation_id", conv.ConversationID). - Msg("Failed to ensure backfill task exists for new room") - } else { - tc.connector.br.WakeupBackfillQueue() - } - } - } else { - if shouldEmitChatInfoUpdate(chatInfo, portal.RoomType) { - tc.userLogin.QueueRemoteEvent(&simplevent.ChatInfoChange{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventChatInfoChange, - PortalKey: portal.PortalKey, - Timestamp: time.Now(), - }, - ChatInfoChange: &bridgev2.ChatInfoChange{ - ChatInfo: chatInfo, - }, - }) + } else if chatInfo.CanBackfill || shouldEmitChatInfoUpdate(chatInfo, portal.RoomType) { + resync := tc.queueChatResyncNow(chatResyncUpdate, portal.PortalKey, chatInfo) + if !resync.Success { + log.Warn(). + Str("conversation_id", conv.ConversationID). + Err(resync.Error). + Msg("Failed to queue ChatResync for existing XChat conversation") } } @@ -519,39 +486,23 @@ func (tc *TwitterClient) syncUntrustedConversation(ctx context.Context, conv *ty chatInfo := tc.conversationToChatInfo(ctx, conv, inbox) - // Create Matrix room if it doesn't exist + // Queue a ChatResync so bridgev2 owns room creation and backfill task registration. if portal.MXID == "" { - // FIXME this is wrong, CreateMatrixRoom should not be called manually - err = portal.CreateMatrixRoom(ctx, tc.userLogin, chatInfo) - if err != nil { - log.Warn().Err(err). + resync := tc.queueChatResyncNow(chatResyncCreate, portal.PortalKey, chatInfo) + if !resync.Success { + log.Warn(). Str("conversation_id", conv.ConversationID). - Msg("Failed to create Matrix room for untrusted conversation") + Err(resync.Error). + Msg("Failed to queue ChatResync for untrusted conversation") return } } else { - // Room already exists - update MessageRequest status via ChatInfoChange - tc.userLogin.QueueRemoteEvent(&simplevent.ChatInfoChange{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventChatInfoChange, - PortalKey: portal.PortalKey, - Timestamp: time.Now(), - }, - ChatInfoChange: &bridgev2.ChatInfoChange{ - ChatInfo: chatInfo, - }, - }) - } - - // Ensure untrusted conversations also have a queue backfill task once a room exists. - if portal.MXID != "" && chatInfo.CanBackfill { - // FIXME this is wrong, backfill tasks are created automatically based on chat resyncs - if err := tc.connector.br.DB.BackfillTask.EnsureExists(ctx, portal.PortalKey, tc.userLogin.ID); err != nil { - log.Warn().Err(err). + resync := tc.queueChatResyncNow(chatResyncUpdate, portal.PortalKey, chatInfo) + if !resync.Success { + log.Warn(). Str("conversation_id", conv.ConversationID). - Msg("Failed to ensure backfill task exists for untrusted conversation") - } else { - tc.connector.br.WakeupBackfillQueue() + Err(resync.Error). + Msg("Failed to queue ChatResync for existing untrusted conversation") } } @@ -586,7 +537,7 @@ func (tc *TwitterClient) processUntrustedMessages(ctx context.Context, conversat } // Queue the message event - tc.HandlePollingEvent(msg, inbox) + tc.HandlePollingEvent(ctx, msg, inbox) } } diff --git a/pkg/connector/client.go b/pkg/connector/client.go index f187309..e6e2094 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -81,7 +81,10 @@ func NewTwitterClient(login *bridgev2.UserLogin, connector *TwitterConnector, cl if !ok { return displayname } - ghost, err := tc.connector.br.GetGhostByID(context.TODO(), userID) + if ctx.Ctx == nil { + return displayname + } + ghost, err := tc.connector.br.GetGhostByID(ctx.Ctx, userID) if err != nil || len(ghost.Identifiers) < 1 { return displayname } diff --git a/pkg/connector/conversationdata.go b/pkg/connector/conversationdata.go index a811c52..693d038 100644 --- a/pkg/connector/conversationdata.go +++ b/pkg/connector/conversationdata.go @@ -112,7 +112,7 @@ func (tc *TwitterClient) ensurePortalForConversation(ctx context.Context, conver log.Warn().Err(err).Msg("Failed to process key change events for fetched conversation data") } - // Sync channel (creates portal if needed) + // Sync channel and queue a resync if the portal still needs a room. tc.syncXChatChannel(ctx, item, users) // Process messages/read events to backfill and register any keys embedded there diff --git a/pkg/connector/handlematrix.go b/pkg/connector/handlematrix.go index 6682cc8..1435e0e 100644 --- a/pkg/connector/handlematrix.go +++ b/pkg/connector/handlematrix.go @@ -590,7 +590,11 @@ func (tc *TwitterClient) doHandleMatrixReaction(ctx context.Context, remove bool // XChat reactions are sent as encrypted MessageCreateEvents (reaction_add/reaction_remove). xchatConvID := NormalizeConversationID(conversationID) - _, err := tc.client.SendEncryptedReaction(ctx, xchatConvID, messageID, emoji, remove) + action := twittermeow.SendEncryptedReactionAdd + if remove { + action = twittermeow.SendEncryptedReactionRemove + } + _, err := tc.client.SendEncryptedReaction(ctx, xchatConvID, messageID, emoji, action) return err } @@ -781,7 +785,7 @@ func (tc *TwitterClient) HandleMatrixViewingChat(ctx context.Context, chat *brid if chat.Portal != nil { conversationID = ParsePortalID(chat.Portal.ID) } - tc.client.SetActiveConversation(ConvertConversationIDToREST(conversationID)) + tc.client.SetActiveConversation(context.WithoutCancel(ctx), ConvertConversationIDToREST(conversationID)) return nil } diff --git a/pkg/connector/handletwit.go b/pkg/connector/handletwit.go index 28aaa3d..f753144 100644 --- a/pkg/connector/handletwit.go +++ b/pkg/connector/handletwit.go @@ -88,6 +88,51 @@ func (tc *TwitterClient) HandleStreamEvent(evt response.StreamEvent) { } } +type chatResyncMode uint8 + +const ( + chatResyncUpdate chatResyncMode = iota + chatResyncCreate + chatResyncCreateWithBackfill +) + +func (tc *TwitterClient) queueChatResync( + mode chatResyncMode, + portalKey networkid.PortalKey, + chatInfo *bridgev2.ChatInfo, + timestamp time.Time, + streamOrder int64, +) bridgev2.EventHandlingResult { + evt := &simplevent.ChatResync{ + EventMeta: simplevent.EventMeta{ + Type: bridgev2.RemoteEventChatResync, + PortalKey: portalKey, + Timestamp: timestamp, + StreamOrder: streamOrder, + }, + ChatInfo: chatInfo, + } + + switch mode { + case chatResyncUpdate: + case chatResyncCreate: + evt.CreatePortal = true + case chatResyncCreateWithBackfill: + evt.CreatePortal = true + evt.CheckNeedsBackfillFunc = func(context.Context, *database.Message) (bool, error) { + return true, nil + } + default: + panic("unknown chat resync mode") + } + + return tc.userLogin.QueueRemoteEvent(evt) +} + +func (tc *TwitterClient) queueChatResyncNow(mode chatResyncMode, portalKey networkid.PortalKey, chatInfo *bridgev2.ChatInfo) bridgev2.EventHandlingResult { + return tc.queueChatResync(mode, portalKey, chatInfo, time.Now(), 0) +} + // buildMemberChangeEvent creates a ChatInfoChange event for participant joins/leaves. func (tc *TwitterClient) buildMemberChangeEvent( conversationID, eventID, eventTime string, @@ -152,7 +197,8 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit targetMessageID = eventID } - if ctx == nil || ctx.Value(ensurePortalContextKey{}) == nil { + bootstrapCreatePortal := ctx.Value(ensurePortalContextKey{}) != nil + if !bootstrapCreatePortal { if _, err := tc.ensurePortalForConversation(ctx, evt.ConversationID, requiredKeyVersion); err != nil { log.Warn(). Err(err). @@ -178,7 +224,7 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit Bool("is_from_me", isFromMe) }, PortalKey: portalKey, - CreatePortal: false, + CreatePortal: bootstrapCreatePortal, Sender: tc.MakeEventSender(evt.MessageData.SenderID), StreamOrder: streamOrder, Timestamp: methods.ParseMsecTimestamp(evt.Time), @@ -206,7 +252,8 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit streamOrder = methods.ParseInt64(evt.Time) } - if ctx == nil || ctx.Value(ensurePortalContextKey{}) == nil { + bootstrapCreatePortal := ctx.Value(ensurePortalContextKey{}) != nil + if !bootstrapCreatePortal { if _, err := tc.ensurePortalForConversation(ctx, evt.ConversationID, requiredKeyVersion); err != nil { log.Warn(). Err(err). @@ -233,7 +280,7 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit Bool("is_from_me", isFromMe) }, PortalKey: portalKey, - CreatePortal: false, // Portal should already exist from initial sync + CreatePortal: bootstrapCreatePortal, Sender: tc.MakeEventSender(evt.MessageData.SenderID), StreamOrder: streamOrder, Timestamp: methods.ParseMsecTimestamp(evt.Time), @@ -361,10 +408,6 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit return tc.userLogin.QueueRemoteEvent(portalDeleteRemoteEvent).Success case *types.ConversationNameUpdate: - if ctx == nil { - ctx = context.TODO() - } - // XChat group titles are encrypted. Decrypt before forwarding to Matrix so // we don't set the room name to ciphertext. newName := evt.ConversationName @@ -479,19 +522,13 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit return false } - return tc.userLogin.QueueRemoteEvent(&simplevent.ChatResync{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventChatResync, - PortalKey: portalKey, - CreatePortal: true, - Timestamp: methods.ParseMsecTimestamp(evt.Time), - StreamOrder: methods.ParseInt64(evt.ID), - }, - ChatInfo: chatInfo, - CheckNeedsBackfillFunc: func(ctx context.Context, latestMessage *database.Message) (bool, error) { - return true, nil - }, - }).Success + return tc.queueChatResync( + chatResyncCreateWithBackfill, + portalKey, + chatInfo, + methods.ParseMsecTimestamp(evt.Time), + methods.ParseInt64(evt.ID), + ).Success } return true @@ -530,16 +567,13 @@ func (tc *TwitterClient) HandleXChatEvent(ctx context.Context, rawEvt types.Twit return false } - return tc.userLogin.QueueRemoteEvent(&simplevent.ChatResync{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventChatResync, - PortalKey: tc.MakePortalKeyFromID(evt.ConversationID), - CreatePortal: true, - Timestamp: methods.ParseMsecTimestamp(evt.Time), - StreamOrder: methods.ParseInt64(evt.ID), - }, - ChatInfo: chatInfo, - }).Success + return tc.queueChatResync( + chatResyncCreate, + tc.MakePortalKeyFromID(evt.ConversationID), + chatInfo, + methods.ParseMsecTimestamp(evt.Time), + methods.ParseInt64(evt.ID), + ).Success default: log.Debug(). @@ -557,10 +591,10 @@ var _ = payload.FailureType(0) // This is used for untrusted (message request) conversations that don't // receive real-time updates via XChat WebSocket. // Returns true to continue polling, false to stop. -func (tc *TwitterClient) HandlePollingEvent(evt types.TwitterEvent, inbox *response.TwitterInboxData) bool { +func (tc *TwitterClient) HandlePollingEvent(ctx context.Context, evt types.TwitterEvent, inbox *response.TwitterInboxData) bool { // Always cache users from inbox when available - needed for portal creation if inbox != nil { - tc.updateTwitterUserInfo(context.TODO(), inbox) + tc.updateTwitterUserInfo(ctx, inbox) tc.userCacheLock.Lock() for userID, user := range inbox.Users { tc.userCache[userID] = user @@ -615,7 +649,6 @@ func (tc *TwitterClient) HandlePollingEvent(evt types.TwitterEvent, inbox *respo // Skip if conversation can use XChat (has encryption keys) - XChat WebSocket handles those portalKey := tc.MakePortalKeyFromID(conversationID) - ctx := context.TODO() if portal, err := tc.connector.br.GetPortalByKey(ctx, portalKey); err == nil && portal != nil { meta := portal.Metadata.(*PortalMetadata) if meta.CanUseXChat() { @@ -629,7 +662,7 @@ func (tc *TwitterClient) HandlePollingEvent(evt types.TwitterEvent, inbox *respo // Dispatch to the appropriate handler based on event type switch e := evt.(type) { case *types.Message: - return tc.handlePollingMessage(e, inbox) + return tc.handlePollingMessage(ctx, e, inbox) case *types.MessageReactionCreate: reaction := (*types.MessageReaction)(e) portalKey := tc.MakePortalKeyFromID(conversationID) @@ -681,7 +714,6 @@ func (tc *TwitterClient) HandlePollingEvent(evt types.TwitterEvent, inbox *respo Msg("Conversation became trusted via polling") // Update portal metadata to mark as trusted (same as XChat path) - ctx := context.TODO() portalKey := tc.MakePortalKeyFromID(conversationID) portal, err := tc.connector.br.GetPortalByKey(ctx, portalKey) if err != nil { @@ -705,16 +737,13 @@ func (tc *TwitterClient) HandlePollingEvent(evt types.TwitterEvent, inbox *respo return false } - return tc.userLogin.QueueRemoteEvent(&simplevent.ChatResync{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventChatResync, - PortalKey: tc.MakePortalKeyFromID(conversationID), - CreatePortal: true, - Timestamp: methods.ParseMsecTimestamp(e.Time), - StreamOrder: methods.ParseInt64(e.ID), - }, - ChatInfo: chatInfo, - }).Success + return tc.queueChatResync( + chatResyncCreate, + tc.MakePortalKeyFromID(conversationID), + chatInfo, + methods.ParseMsecTimestamp(e.Time), + methods.ParseInt64(e.ID), + ).Success } return true @@ -752,7 +781,7 @@ func (tc *TwitterClient) markPollingChatResyncSuccess(conversationID string, now } // handlePollingMessage handles a message event from REST API polling. -func (tc *TwitterClient) handlePollingMessage(evt *types.Message, inbox *response.TwitterInboxData) bool { +func (tc *TwitterClient) handlePollingMessage(ctx context.Context, evt *types.Message, inbox *response.TwitterInboxData) bool { isFromMe := MakeUserLoginID(evt.MessageData.SenderID) == tc.userLogin.ID portalKey := tc.MakePortalKeyFromID(evt.ConversationID) msgID := evt.ID @@ -763,7 +792,6 @@ func (tc *TwitterClient) handlePollingMessage(evt *types.Message, inbox *respons Logger() // For polling messages, ensure the portal exists - ctx := context.TODO() portal, err := tc.connector.br.GetPortalByKey(ctx, portalKey) if err != nil { log.Warn(). @@ -772,7 +800,7 @@ func (tc *TwitterClient) handlePollingMessage(evt *types.Message, inbox *respons return false } - // Create portal if it doesn't exist + // Queue a portal resync if the room doesn't exist yet. if portal.MXID == "" { chatInfo := tc.getOrFetchChatInfoForPolling(ctx, evt.ConversationID, inbox) if chatInfo == nil { @@ -780,21 +808,19 @@ func (tc *TwitterClient) handlePollingMessage(evt *types.Message, inbox *respons Msg("Failed to get chat info for polling message") return false } - // FIXME this is wrong, CreateMatrixRoom should not be called manually - if err := portal.CreateMatrixRoom(ctx, tc.userLogin, chatInfo); err != nil { + resync := tc.queueChatResync( + chatResyncCreate, + portal.PortalKey, + chatInfo, + methods.ParseMsecTimestamp(evt.Time), + methods.ParseInt64(msgID), + ) + if !resync.Success { log.Warn(). - Err(err). - Msg("Failed to create Matrix room for polling message") + Err(resync.Error). + Msg("Failed to queue ChatResync for polling message") return false } - // Register backfill task for the newly created room - // FIXME this is wrong, backfill tasks are created automatically based on chat resyncs - if err := tc.connector.br.DB.BackfillTask.EnsureExists(ctx, portal.PortalKey, tc.userLogin.ID); err != nil { - log.Warn().Err(err). - Msg("Failed to ensure backfill task exists for new polling room") - } else { - tc.connector.br.WakeupBackfillQueue() - } } else if tc.shouldAttemptPollingChatResync(evt.ConversationID, now) { chatInfo := tc.getOrFetchChatInfoForPolling(ctx, evt.ConversationID, inbox) if !tc.isDMChatInfoComplete(chatInfo) { @@ -806,21 +832,18 @@ func (tc *TwitterClient) handlePollingMessage(evt *types.Message, inbox *respons Int("missing_userinfo", missingUserInfo). Msg("Skipping polling ChatResync: incomplete chat info") } else { - ok := tc.userLogin.QueueRemoteEvent(&simplevent.ChatResync{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventChatResync, - PortalKey: portal.PortalKey, - CreatePortal: true, - Timestamp: methods.ParseMsecTimestamp(evt.Time), - StreamOrder: methods.ParseInt64(msgID), - }, - ChatInfo: chatInfo, - }).Success - if ok { + resync := tc.queueChatResync( + chatResyncUpdate, + portal.PortalKey, + chatInfo, + methods.ParseMsecTimestamp(evt.Time), + methods.ParseInt64(msgID), + ) + if resync.Success { log.Debug().Msg("Queued polling ChatResync") tc.markPollingChatResyncSuccess(evt.ConversationID, now) } else { - log.Debug().Msg("Failed to queue polling ChatResync") + log.Debug().Err(resync.Error).Msg("Failed to queue polling ChatResync") } } } diff --git a/pkg/twittermeow/account.go b/pkg/twittermeow/account.go index 9041920..d464ea1 100644 --- a/pkg/twittermeow/account.go +++ b/pkg/twittermeow/account.go @@ -40,36 +40,41 @@ func (c *Client) GetCurrentUserProfile(ctx context.Context) (CurrentUserProfile, if err != nil { return CurrentUserProfile{}, err } - if len(resp.Errors) > 0 && resp.Errors[0].Message != "" { - return CurrentUserProfile{}, fmt.Errorf("GetUsersByIdsForXChat error: %s", resp.Errors[0].Message) + if len(resp.Errors) > 0 { + msg := strings.TrimSpace(resp.Errors[0].Message) + if msg == "" { + msg = "unknown error" + } + return CurrentUserProfile{}, fmt.Errorf("get user profile: %s", msg) } if len(resp.Data.GetMemberResults.Results) != 1 { return CurrentUserProfile{}, fmt.Errorf("expected 1 user result for %s, got %d", currentUserID, len(resp.Data.GetMemberResults.Results)) } result := resp.Data.GetMemberResults.Results[0] - if result.MemberResults == nil || result.MemberResults.Result == nil || result.MemberResults.Result.Core == nil { + member := result.MemberResults + if member == nil || member.Result == nil || member.Result.Core == nil { return CurrentUserProfile{}, fmt.Errorf("GetUsersByIdsForXChat returned no user for %s", currentUserID) } - resultUserID := result.MemberResults.RestID + resultUserID := member.RestID + if resultUserID == "" { + resultUserID = member.Result.RestID + } if resultUserID == "" { - resultUserID = result.MemberResults.Result.RestID + resultUserID = currentUserID } - if resultUserID != "" && resultUserID != currentUserID { + if resultUserID != currentUserID { return CurrentUserProfile{}, fmt.Errorf("GetUsersByIdsForXChat returned user %s for %s", resultUserID, currentUserID) } profile := CurrentUserProfile{ - ID: currentUserID, - ScreenName: strings.TrimSpace(result.MemberResults.Result.Core.ScreenName), - Name: strings.TrimSpace(result.MemberResults.Result.Core.Name), - } - if resultUserID != "" { - profile.ID = resultUserID + ID: resultUserID, + ScreenName: strings.TrimSpace(member.Result.Core.ScreenName), + Name: strings.TrimSpace(member.Result.Core.Name), } - if result.MemberResults.Result.Avatar != nil { - profile.AvatarURL = strings.TrimSpace(result.MemberResults.Result.Avatar.ImageURL) + if member.Result.Avatar != nil { + profile.AvatarURL = strings.TrimSpace(member.Result.Avatar.ImageURL) } return profile, nil diff --git a/pkg/twittermeow/client.go b/pkg/twittermeow/client.go index 54c13aa..6144a8c 100644 --- a/pkg/twittermeow/client.go +++ b/pkg/twittermeow/client.go @@ -27,7 +27,7 @@ import ( "go.mau.fi/mautrix-twitter/pkg/twittermeow/methods" ) -type EventHandler func(evt types.TwitterEvent, inbox *response.TwitterInboxData) bool +type EventHandler func(ctx context.Context, evt types.TwitterEvent, inbox *response.TwitterInboxData) bool type StreamEventHandler func(evt response.StreamEvent) // ConversationDataCallback is called when conversation data is refreshed (e.g., during key refresh). @@ -480,8 +480,8 @@ func (c *Client) makeAPIRequest(ctx context.Context, apiRequestOpts apiRequestOp return c.MakeRequest(ctx, apiRequestOpts.URL, apiRequestOpts.Method, headers, apiRequestOpts.Body, apiRequestOpts.ContentType) } -func (c *Client) SetActiveConversation(conversationID string) { - c.stream.startOrUpdateEventStream(conversationID) +func (c *Client) SetActiveConversation(ctx context.Context, conversationID string) { + c.stream.startOrUpdateEventStream(ctx, conversationID) } // FetchRaw performs an authenticated request to the given URL and returns the response and body. diff --git a/pkg/twittermeow/messaging.go b/pkg/twittermeow/messaging.go index 82bc699..5eda773 100644 --- a/pkg/twittermeow/messaging.go +++ b/pkg/twittermeow/messaging.go @@ -414,6 +414,13 @@ type SendEncryptedEditOpts struct { Entities []*payload.RichTextEntity } +type SendEncryptedReactionAction uint8 + +const ( + SendEncryptedReactionAdd SendEncryptedReactionAction = iota + SendEncryptedReactionRemove +) + func (c *Client) sendMessageMutation(ctx context.Context, pl *payload.SendMessageMutationPayload) (*response.SendMessageMutationResponse, error) { token, err := c.ensureConversationToken(ctx, pl.Variables.ConversationID) if err != nil { @@ -459,22 +466,20 @@ func (c *Client) sendMessageMutation(ctx context.Context, pl *payload.SendMessag // SendEncryptedReaction sends a reaction add/remove via the XChat protocol. // targetMessageSequenceID must be the XChat message sequence ID of the message being reacted to. -func (c *Client) SendEncryptedReaction(ctx context.Context, conversationID, targetMessageSequenceID, emoji string, remove bool) (*response.SendMessageMutationResponse, error) { - token, err := c.ensureConversationToken(ctx, conversationID) - if err != nil { - return nil, fmt.Errorf("get conversation token: %w", err) - } - +func (c *Client) SendEncryptedReaction(ctx context.Context, conversationID, targetMessageSequenceID, emoji string, action SendEncryptedReactionAction) (*response.SendMessageMutationResponse, error) { messageID := uuid.NewString() builder := crypto.NewMessageBuilder(c.keyManager, c.GetCurrentUserID()). SetMessageID(messageID). SetConversationID(conversationID) - if remove { - builder.SetReactionRemove(targetMessageSequenceID, emoji) - } else { + switch action { + case SendEncryptedReactionAdd: builder.SetReactionAdd(targetMessageSequenceID, emoji) + case SendEncryptedReactionRemove: + builder.SetReactionRemove(targetMessageSequenceID, emoji) + default: + panic("unknown encrypted reaction action") } encodedMCE, encodedSig, err := builder.BuildForSend(ctx) @@ -490,7 +495,6 @@ func (c *Client) SendEncryptedReaction(ctx context.Context, conversationID, targ pl := payload.NewSendMessageMutationPayload(payload.SendMessageMutationVariables{ ConversationID: conversationID, MessageID: messageID, - ConversationToken: token, EncodedMessageCreateEvent: encodedMCE, EncodedMessageEventSignature: sigPtr, }) @@ -508,11 +512,6 @@ func (c *Client) SendEncryptedEdit(ctx context.Context, opts SendEncryptedEditOp return nil, fmt.Errorf("target message sequence ID is required") } - token, err := c.ensureConversationToken(ctx, opts.ConversationID) - if err != nil { - return nil, fmt.Errorf("get conversation token: %w", err) - } - messageID := opts.MessageID if messageID == "" { messageID = uuid.NewString() @@ -536,7 +535,6 @@ func (c *Client) SendEncryptedEdit(ctx context.Context, opts SendEncryptedEditOp pl := payload.NewSendMessageMutationPayload(payload.SendMessageMutationVariables{ ConversationID: opts.ConversationID, MessageID: messageID, - ConversationToken: token, EncodedMessageCreateEvent: encodedMCE, EncodedMessageEventSignature: sigPtr, }) @@ -546,12 +544,6 @@ func (c *Client) SendEncryptedEdit(ctx context.Context, opts SendEncryptedEditOp // SendEncryptedMessage sends an encrypted message via the XChat protocol. func (c *Client) SendEncryptedMessage(ctx context.Context, opts SendEncryptedMessageOpts) (*response.SendMessageMutationResponse, error) { - // Get the server-provided conversation token for this conversation - token, err := c.ensureConversationToken(ctx, opts.ConversationID) - if err != nil { - return nil, fmt.Errorf("get conversation token: %w", err) - } - messageID := opts.MessageID if messageID == "" { messageID = uuid.NewString() @@ -599,7 +591,6 @@ func (c *Client) SendEncryptedMessage(ctx context.Context, opts SendEncryptedMes pl := payload.NewSendMessageMutationPayload(payload.SendMessageMutationVariables{ ConversationID: opts.ConversationID, MessageID: messageID, - ConversationToken: token, EncodedMessageCreateEvent: encodedMCE, EncodedMessageEventSignature: sigPtr, }) diff --git a/pkg/twittermeow/methods/html_test.go b/pkg/twittermeow/methods/html_test.go index 8385d34..6cf0d26 100644 --- a/pkg/twittermeow/methods/html_test.go +++ b/pkg/twittermeow/methods/html_test.go @@ -8,6 +8,34 @@ import ( ) func TestParseOndemandSURLFromScript(t *testing.T) { + tests := []struct { + name string + js string + want string + }{ + { + name: "find url", + js: `123:"ondemand.s",{123:"deadbeef"}`, + want: "https://abs.twimg.com/responsive-web/client-web/ondemand.s.deadbeefa.js", + }, + { + name: "missing chunk", + js: `123:"main",{123:"deadbeef"}`, + want: "", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := ParseOndemandSURLFromScript([]byte(test.js)) + if got != test.want { + t.Fatalf("unexpected ondemand url: got %q want %q", got, test.want) + } + }) + } +} + +func TestParseOndemandSURLFromLiveXBootstrap(t *testing.T) { client := &http.Client{Timeout: 20 * time.Second} req, err := http.NewRequest(http.MethodGet, "https://x.com/", nil) if err != nil { diff --git a/pkg/twittermeow/polling.go b/pkg/twittermeow/polling.go index e920b05..0effbc5 100644 --- a/pkg/twittermeow/polling.go +++ b/pkg/twittermeow/polling.go @@ -76,7 +76,7 @@ func (pc *PollingClient) doPoll(ctx context.Context) { } else if err != nil { log.Err(err).Msg("Failed to poll for updates") authError := IsAuthError(err) - pc.client.eventHandler(&types.PollingError{Error: err, IsAuth: authError}, nil) + pc.client.eventHandler(ctx, &types.PollingError{Error: err, IsAuth: authError}, nil) if authError { return } @@ -88,7 +88,7 @@ func (pc *PollingClient) doPoll(ctx context.Context) { tick.Reset(backoffInterval) } else if failing { failing = false - pc.client.eventHandler(&types.PollingError{}, nil) + pc.client.eventHandler(ctx, &types.PollingError{}, nil) backoffInterval = defaultPollingInterval / 2 tick.Reset(defaultPollingInterval) } @@ -119,7 +119,7 @@ func (pc *PollingClient) poll(ctx context.Context) error { return nil } - if !pc.client.eventHandler(nil, userUpdatesResponse.UserEvents) { + if !pc.client.eventHandler(ctx, nil, userUpdatesResponse.UserEvents) { return errEventHandlerFailed } for _, entry := range userUpdatesResponse.UserEvents.Entries { @@ -128,7 +128,7 @@ func (pc *PollingClient) poll(ctx context.Context) error { } parsed := entry.ParseWithErrorLog(&pc.client.Logger) if parsed != nil { - if !pc.client.eventHandler(parsed, userUpdatesResponse.UserEvents) { + if !pc.client.eventHandler(ctx, parsed, userUpdatesResponse.UserEvents) { return errEventHandlerFailed } } diff --git a/pkg/twittermeow/stream_client.go b/pkg/twittermeow/stream_client.go index 3654256..61c37a5 100644 --- a/pkg/twittermeow/stream_client.go +++ b/pkg/twittermeow/stream_client.go @@ -42,8 +42,8 @@ func (c *Client) newStreamClient() *StreamClient { return sc } -func (sc *StreamClient) startOrUpdateEventStream(conversationID string) { - ctx := sc.client.Logger.With().Str("action", "event stream").Logger().WithContext(context.TODO()) +func (sc *StreamClient) startOrUpdateEventStream(ctx context.Context, conversationID string) { + ctx = sc.client.Logger.With().Str("action", "event stream").Logger().WithContext(ctx) if sc.conversationID == "" { sc.conversationID = conversationID go sc.start(ctx) diff --git a/pkg/twittermeow/xchat_send.go b/pkg/twittermeow/xchat_send.go index 4506ff6..8cf0726 100644 --- a/pkg/twittermeow/xchat_send.go +++ b/pkg/twittermeow/xchat_send.go @@ -87,11 +87,12 @@ func (c *Client) ensureConversationToken(ctx context.Context, conversationID str if err == nil { return token, nil } - if err != nil && !errors.Is(err, crypto.ErrKeyNotFound) { + if !errors.Is(err, crypto.ErrKeyNotFound) { return "", fmt.Errorf("get conversation token: %w", err) } - if err := c.refreshConversationToken(ctx, conversationID); err != nil && !errors.Is(err, crypto.ErrKeyNotFound) { + err = c.refreshConversationToken(ctx, conversationID) + if err != nil { return "", err } @@ -125,33 +126,32 @@ func (c *Client) refreshConversationToken(ctx context.Context, conversationID st } for _, encodedEvt := range encoded { - err := c.putConversationTokenFromEncodedEvent(ctx, conversationID, encodedEvt) - if err == nil { - return nil - } - if !errors.Is(err, crypto.ErrKeyNotFound) { - return err + tokenConversationID, token := conversationTokenFromEncodedEvent(conversationID, encodedEvt) + if token == "" { + continue } + return c.keyManager.PutConversationToken(ctx, tokenConversationID, token) } return crypto.ErrKeyNotFound } -func (c *Client) putConversationTokenFromEncodedEvent(ctx context.Context, conversationID, encoded string) error { +func conversationTokenFromEncodedEvent(fallbackConversationID, encoded string) (conversationID, token string) { if encoded == "" { - return crypto.ErrKeyNotFound + return "", "" } evt, err := DecodeMessageEvent(encoded) if err != nil { - return crypto.ErrKeyNotFound + return "", "" } if evt == nil || evt.ConversationToken == nil || *evt.ConversationToken == "" { - return crypto.ErrKeyNotFound + return "", "" } + conversationID = fallbackConversationID if evt.ConversationId != nil && *evt.ConversationId != "" { conversationID = *evt.ConversationId } - return c.keyManager.PutConversationToken(ctx, conversationID, *evt.ConversationToken) + return conversationID, *evt.ConversationToken } // getSelfConversationID returns the user's self-conversation ID (user_id:user_id format). @@ -168,11 +168,6 @@ func (c *Client) SendXChatPinConversation(ctx context.Context, targetConversatio selfConvID := c.getSelfConversationID() - token, err := c.ensureConversationToken(ctx, selfConvID) - if err != nil { - return fmt.Errorf("get self conversation token: %w", err) - } - messageID := uuid.NewString() builder := crypto.NewMessageBuilder(c.keyManager, c.GetCurrentUserID()). @@ -193,7 +188,6 @@ func (c *Client) SendXChatPinConversation(ctx context.Context, targetConversatio pl := payload.NewSendMessageMutationPayload(payload.SendMessageMutationVariables{ ConversationID: selfConvID, MessageID: messageID, - ConversationToken: token, EncodedMessageCreateEvent: encodedMCE, EncodedMessageEventSignature: sigPtr, }) @@ -210,11 +204,6 @@ func (c *Client) SendXChatUnpinConversation(ctx context.Context, targetConversat selfConvID := c.getSelfConversationID() - token, err := c.ensureConversationToken(ctx, selfConvID) - if err != nil { - return fmt.Errorf("get self conversation token: %w", err) - } - messageID := uuid.NewString() builder := crypto.NewMessageBuilder(c.keyManager, c.GetCurrentUserID()). @@ -235,7 +224,6 @@ func (c *Client) SendXChatUnpinConversation(ctx context.Context, targetConversat pl := payload.NewSendMessageMutationPayload(payload.SendMessageMutationVariables{ ConversationID: selfConvID, MessageID: messageID, - ConversationToken: token, EncodedMessageCreateEvent: encodedMCE, EncodedMessageEventSignature: sigPtr, })