From 2963ffb7ad9900c111c2c0c0c51fe9c811784382 Mon Sep 17 00:00:00 2001 From: highesttt Date: Tue, 28 Jul 2026 12:27:08 -0400 Subject: [PATCH 1/4] fix: allow reusing LINE custom reactions --- pkg/connector/client.go | 1 - pkg/connector/connector.go | 10 +- pkg/connector/reaction.go | 197 ++++++++++++++++++++++++++----- pkg/connector/reaction_test.go | 208 +++++++++++++++++++++++++++++++++ pkg/connector/sync.go | 154 +++++++++--------------- 5 files changed, 435 insertions(+), 135 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index ee0c355..4415b17 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -80,7 +80,6 @@ type LineClient struct { knownMemberChatMIDs map[string]struct{} // chatMid -> current member chats returned by getAllChatMids reactionIconMXC map[int]string // predefinedReactionType -> cached MXC URI paidReactionIconMXC map[string]string // LINE sticon URL -> cached MXC URI - recentReactions sync.Map // "msgID\x00emoji" -> struct{} to dedup concurrent 139/140 events unblockBackfills sync.Map // chat MID -> *unblockBackfillState while unblock history restoration is active wg sync.WaitGroup diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 0fdced6..54cd1cf 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -99,10 +99,12 @@ func (lc *LineConnector) GetConfig() (example string, data any, upgrader configu func (lc *LineConnector) GetDBMetaTypes() database.MetaTypes { return database.MetaTypes{ - Portal: nil, - Ghost: nil, - Message: nil, - Reaction: nil, + Portal: nil, + Ghost: nil, + Message: nil, + Reaction: func() any { + return &ReactionMetadata{} + }, UserLogin: func() any { return &UserLoginMetadata{} }, diff --git a/pkg/connector/reaction.go b/pkg/connector/reaction.go index b9f2b8b..64661aa 100644 --- a/pkg/connector/reaction.go +++ b/pkg/connector/reaction.go @@ -36,8 +36,13 @@ type linePaidReactionRef struct { Version int } -func (ref linePaidReactionRef) networkEmojiID() networkid.EmojiID { - return networkid.EmojiID("paid:" + ref.ProductID + ":" + ref.EmojiID) +type lineReactionRef struct { + typ line.ReactionType +} + +type ReactionMetadata struct { + MatrixKey string `json:"matrix_key,omitempty"` + ReactionType line.ReactionType `json:"reaction_type"` } func (ref linePaidReactionRef) reactionType() line.ReactionType { @@ -51,6 +56,61 @@ func (ref linePaidReactionRef) reactionType() line.ReactionType { } } +func cloneLineReactionType(typ line.ReactionType) line.ReactionType { + cloned := line.ReactionType{ + PredefinedReactionType: typ.PredefinedReactionType, + } + if typ.PaidReactionType != nil { + paid := *typ.PaidReactionType + cloned.PaidReactionType = &paid + } + return cloned +} + +func newLineReactionRef(typ line.ReactionType) (lineReactionRef, error) { + hasPredefined := typ.PredefinedReactionType != 0 + hasPaid := typ.PaidReactionType != nil + if hasPredefined == hasPaid { + return lineReactionRef{}, errors.New("reaction type must contain exactly one predefined or paid reaction") + } + if hasPredefined { + if _, ok := line.PredefinedReactionEmoji[typ.PredefinedReactionType]; !ok { + return lineReactionRef{}, fmt.Errorf("unknown predefined reaction type %d", typ.PredefinedReactionType) + } + } else if typ.PaidReactionType.ProductID == "" || typ.PaidReactionType.EmojiID == "" { + return lineReactionRef{}, errors.New("paid reaction is missing product or emoji ID") + } + return lineReactionRef{typ: cloneLineReactionType(typ)}, nil +} + +func (ref lineReactionRef) reactionType() line.ReactionType { + return cloneLineReactionType(ref.typ) +} + +func (ref lineReactionRef) networkEmojiID() networkid.EmojiID { + if ref.typ.PaidReactionType != nil { + return networkid.EmojiID("paid:" + ref.typ.PaidReactionType.ProductID + ":" + ref.typ.PaidReactionType.EmojiID) + } + return networkid.EmojiID("predefined:" + strconv.Itoa(ref.typ.PredefinedReactionType)) +} + +func (ref lineReactionRef) equal(other lineReactionRef) bool { + if ref.typ.PredefinedReactionType != other.typ.PredefinedReactionType { + return false + } + if ref.typ.PaidReactionType == nil || other.typ.PaidReactionType == nil { + return ref.typ.PaidReactionType == nil && other.typ.PaidReactionType == nil + } + return *ref.typ.PaidReactionType == *other.typ.PaidReactionType +} + +func (ref lineReactionRef) metadata(matrixKey string) *ReactionMetadata { + return &ReactionMetadata{ + MatrixKey: matrixKey, + ReactionType: ref.reactionType(), + } +} + // These are the LINE emoji/sticon URLs from the issue's pack-based reaction // set. Add more entries here as more Matrix emoji -> LINE CDN URL mappings are // captured. @@ -302,6 +362,36 @@ func (lc *LineClient) getPaidReactionMXC(ctx context.Context, prt *line.PaidReac return mxc, nil } +func (lc *LineClient) convertReaction( + ctx context.Context, + typ line.ReactionType, + sender bridgev2.EventSender, + timestamp time.Time, +) (*bridgev2.BackfillReaction, error) { + ref, err := newLineReactionRef(typ) + if err != nil { + return nil, err + } + + var mxc string + if ref.typ.PaidReactionType != nil { + mxc, err = lc.getPaidReactionMXC(ctx, ref.typ.PaidReactionType) + } else { + mxc, err = lc.getPredefinedReactionMXC(ctx, ref.typ.PredefinedReactionType) + } + if err != nil { + return nil, err + } + + return &bridgev2.BackfillReaction{ + Timestamp: timestamp, + Sender: sender, + EmojiID: ref.networkEmojiID(), + Emoji: mxc, + DBMetadata: ref.metadata(mxc), + }, nil +} + func (lc *LineClient) convertMessageReactions(ctx context.Context, msg *line.Message) ([]*bridgev2.BackfillReaction, bool) { if msg == nil || msg.Reactions == nil { return nil, false @@ -319,18 +409,16 @@ func (lc *LineClient) convertMessageReactions(ctx context.Context, msg *line.Mes continue } - var ( - mxc string - err error - ) - switch { - case reaction.ReactionType.PaidReactionType != nil: - mxc, err = lc.getPaidReactionMXC(ctx, reaction.ReactionType.PaidReactionType) - case reaction.ReactionType.PredefinedReactionType != 0: - mxc, err = lc.getPredefinedReactionMXC(ctx, reaction.ReactionType.PredefinedReactionType) - default: - err = errors.New("reaction type is missing") + var timestamp time.Time + if timestampMillis, err := reaction.AtMillis.Int64(); err == nil && timestampMillis > 0 { + timestamp = time.UnixMilli(timestampMillis) } + convertedReaction, err := lc.convertReaction( + ctx, + reaction.ReactionType, + lc.eventSenderForMID(reaction.FromUserMID), + timestamp, + ) if err != nil { complete = false lc.UserLogin.Bridge.Log.Warn(). @@ -340,16 +428,7 @@ func (lc *LineClient) convertMessageReactions(ctx context.Context, msg *line.Mes Msg("Skipping unsupported historical reaction") continue } - - var timestamp time.Time - if timestampMillis, err := reaction.AtMillis.Int64(); err == nil && timestampMillis > 0 { - timestamp = time.UnixMilli(timestampMillis) - } - converted = append(converted, &bridgev2.BackfillReaction{ - Timestamp: timestamp, - Sender: lc.eventSenderForMID(reaction.FromUserMID), - Emoji: mxc, - }) + converted = append(converted, convertedReaction) } return converted, complete } @@ -457,6 +536,61 @@ func linePaidReactionForMatrixEmoji(key string) (linePaidReactionRef, bool) { return ref, true } +func storedLineReactionForMatrixKey(key string, reactions []*database.Reaction) (lineReactionRef, bool) { + var ( + found lineReactionRef + hasFound bool + ) + for _, reaction := range reactions { + meta, ok := reaction.Metadata.(*ReactionMetadata) + if !ok || meta == nil || meta.MatrixKey != key { + continue + } + ref, err := newLineReactionRef(meta.ReactionType) + if err != nil || (reaction.EmojiID != "" && reaction.EmojiID != ref.networkEmojiID()) { + return lineReactionRef{}, false + } + if hasFound && !found.equal(ref) { + return lineReactionRef{}, false + } + found = ref + hasFound = true + } + return found, hasFound +} + +func (lc *LineClient) resolveMatrixReaction(ctx context.Context, msg *bridgev2.MatrixReaction) (lineReactionRef, error) { + key := msg.Content.RelatesTo.GetAnnotationKey() + if paidRef, ok := linePaidReactionForMatrixEmoji(key); ok { + ref, err := newLineReactionRef(paidRef.reactionType()) + if err != nil { + return lineReactionRef{}, err + } + return ref, nil + } + if !strings.HasPrefix(key, "mxc://") { + return lineReactionRef{}, unsupportedMatrixReactionError(key) + } + if msg.TargetMessage == nil || msg.Portal == nil || msg.Portal.Bridge == nil || msg.Portal.Bridge.DB == nil { + return lineReactionRef{}, errors.New("reaction target database context is missing") + } + + reactions, err := msg.Portal.Bridge.DB.Reaction.GetAllToMessagePart( + ctx, + msg.Portal.Receiver, + msg.TargetMessage.ID, + msg.TargetMessage.PartID, + ) + if err != nil { + return lineReactionRef{}, fmt.Errorf("get target message reactions: %w", err) + } + ref, ok := storedLineReactionForMatrixKey(key, reactions) + if !ok { + return lineReactionRef{}, unsupportedMatrixReactionError(key) + } + return ref, nil +} + func unsupportedMatrixReactionError(key string) error { return bridgev2.WrapErrorInStatus(fmt.Errorf("LINE does not support Matrix reaction %q", key)). WithStatus(event.MessageStatusFail). @@ -572,9 +706,9 @@ func (lc *LineClient) consumeSentReqSeq(reqSeq int) bool { func (lc *LineClient) PreHandleMatrixReaction(ctx context.Context, msg *bridgev2.MatrixReaction) (bridgev2.MatrixReactionPreResponse, error) { key := msg.Content.RelatesTo.GetAnnotationKey() - ref, ok := linePaidReactionForMatrixEmoji(key) - if !ok { - return bridgev2.MatrixReactionPreResponse{}, unsupportedMatrixReactionError(key) + ref, err := lc.resolveMatrixReaction(ctx, msg) + if err != nil { + return bridgev2.MatrixReactionPreResponse{}, err } return bridgev2.MatrixReactionPreResponse{ SenderID: makeUserID(string(lc.UserLogin.ID)), @@ -586,9 +720,9 @@ func (lc *LineClient) PreHandleMatrixReaction(ctx context.Context, msg *bridgev2 func (lc *LineClient) HandleMatrixReaction(ctx context.Context, msg *bridgev2.MatrixReaction) (*database.Reaction, error) { key := msg.Content.RelatesTo.GetAnnotationKey() - ref, ok := linePaidReactionForMatrixEmoji(key) - if !ok { - return nil, unsupportedMatrixReactionError(key) + ref, err := lc.resolveMatrixReaction(ctx, msg) + if err != nil { + return nil, err } targetID, err := parseReactionTargetMessageID(msg.TargetMessage.ID) if err != nil { @@ -610,8 +744,9 @@ func (lc *LineClient) HandleMatrixReaction(ctx context.Context, msg *bridgev2.Ma } return &database.Reaction{ - EmojiID: ref.networkEmojiID(), - Emoji: key, + EmojiID: ref.networkEmojiID(), + Emoji: key, + Metadata: ref.metadata(key), }, nil } diff --git a/pkg/connector/reaction_test.go b/pkg/connector/reaction_test.go index 7746760..0ec1c84 100644 --- a/pkg/connector/reaction_test.go +++ b/pkg/connector/reaction_test.go @@ -31,6 +31,9 @@ func TestCapabilitiesAdvertiseSupportedReactions(t *testing.T) { if caps.ReactionCount != 1 { t.Fatalf("ReactionCount = %d, want 1", caps.ReactionCount) } + if caps.CustomEmojiReactions { + t.Fatal("CustomEmojiReactions must stay disabled because arbitrary Matrix custom emojis are unsupported") + } if len(caps.AllowedReactions) != len(lineEmojiReactionURLs) { t.Fatalf("AllowedReactions has %d entries, want %d", len(caps.AllowedReactions), len(lineEmojiReactionURLs)) } @@ -431,6 +434,155 @@ func TestLinePaidReactionForMatrixEmoji(t *testing.T) { } } +func TestLineReactionRefIdentityAndMetadata(t *testing.T) { + paidType := line.ReactionType{PaidReactionType: &line.PaidReactionType{ + ProductID: "product", + EmojiID: "emoji", + ResourceType: 2, + Version: 7, + }} + paidRef, err := newLineReactionRef(paidType) + if err != nil { + t.Fatal(err) + } + if paidRef.networkEmojiID() != "paid:product:emoji" { + t.Fatalf("paid EmojiID = %q", paidRef.networkEmojiID()) + } + + meta := paidRef.metadata("mxc://line/custom") + paidType.PaidReactionType.Version = 99 + if meta.MatrixKey != "mxc://line/custom" || meta.ReactionType.PaidReactionType.Version != 7 { + t.Fatalf("paid metadata = %#v", meta) + } + + raw, err := json.Marshal(meta) + if err != nil { + t.Fatal(err) + } + var decoded ReactionMetadata + if err = json.Unmarshal(raw, &decoded); err != nil { + t.Fatal(err) + } + decodedRef, err := newLineReactionRef(decoded.ReactionType) + if err != nil { + t.Fatal(err) + } + if !paidRef.equal(decodedRef) { + t.Fatalf("decoded ref = %#v, want %#v", decodedRef, paidRef) + } + + predefinedRef, err := newLineReactionRef(line.ReactionType{PredefinedReactionType: 2}) + if err != nil { + t.Fatal(err) + } + if predefinedRef.networkEmojiID() != "predefined:2" { + t.Fatalf("predefined EmojiID = %q", predefinedRef.networkEmojiID()) + } + + for _, invalid := range []line.ReactionType{ + {}, + {PredefinedReactionType: 1}, + {PaidReactionType: &line.PaidReactionType{}}, + { + PredefinedReactionType: 2, + PaidReactionType: &line.PaidReactionType{ProductID: "product", EmojiID: "emoji"}, + }, + } { + if _, err = newLineReactionRef(invalid); err == nil { + t.Fatalf("invalid reaction type was accepted: %#v", invalid) + } + } + + metaFactory := (&LineConnector{}).GetDBMetaTypes().Reaction + if metaFactory == nil { + t.Fatal("reaction metadata type is not registered") + } + if _, ok := metaFactory().(*ReactionMetadata); !ok { + t.Fatalf("reaction metadata factory returned %T", metaFactory()) + } +} + +func TestStoredLineReactionForMatrixKey(t *testing.T) { + key := "mxc://line/custom" + paidType := line.ReactionType{PaidReactionType: &line.PaidReactionType{ + ProductID: "product", + EmojiID: "emoji", + ResourceType: 2, + Version: 7, + }} + valid := &database.Reaction{ + EmojiID: "paid:product:emoji", + Metadata: &ReactionMetadata{ + MatrixKey: key, + ReactionType: paidType, + }, + } + same := &database.Reaction{ + EmojiID: "paid:product:emoji", + Metadata: &ReactionMetadata{ + MatrixKey: key, + ReactionType: cloneLineReactionType(paidType), + }, + } + unrelated := &database.Reaction{ + EmojiID: "predefined:2", + Metadata: &ReactionMetadata{ + MatrixKey: "mxc://line/other", + ReactionType: line.ReactionType{PredefinedReactionType: 2}, + }, + } + + ref, ok := storedLineReactionForMatrixKey(key, []*database.Reaction{valid, same, unrelated}) + if !ok || ref.networkEmojiID() != "paid:product:emoji" { + t.Fatalf("stored reaction = %#v, %v", ref, ok) + } + if _, ok = storedLineReactionForMatrixKey("mxc://line/arbitrary", []*database.Reaction{valid}); ok { + t.Fatal("arbitrary MXC was accepted") + } + if _, ok = storedLineReactionForMatrixKey(key, []*database.Reaction{{ + Emoji: key, + }}); ok { + t.Fatal("legacy reaction without metadata was accepted") + } + if _, ok = storedLineReactionForMatrixKey(key, []*database.Reaction{valid, { + EmojiID: "predefined:2", + Metadata: &ReactionMetadata{ + MatrixKey: key, + ReactionType: line.ReactionType{PredefinedReactionType: 2}, + }, + }}); ok { + t.Fatal("conflicting LINE reaction metadata was accepted") + } + if _, ok = storedLineReactionForMatrixKey(key, []*database.Reaction{{ + EmojiID: "paid:different:id", + Metadata: valid.Metadata, + }}); ok { + t.Fatal("reaction metadata with a mismatched stable ID was accepted") + } +} + +func TestPreHandleMatrixReactionKeepsUnicodeBehavior(t *testing.T) { + lc := &LineClient{UserLogin: &bridgev2.UserLogin{ + UserLogin: &database.UserLogin{ID: "Uself"}, + }} + msg := &bridgev2.MatrixReaction{ + MatrixEventBase: bridgev2.MatrixEventBase[*event.ReactionEventContent]{ + Content: &event.ReactionEventContent{RelatesTo: event.RelatesTo{ + Type: event.RelAnnotation, + Key: "\U0001F44D\uFE0F", + }}, + }, + } + + resp, err := lc.PreHandleMatrixReaction(context.Background(), msg) + if err != nil { + t.Fatal(err) + } + if resp.SenderID != "Uself" || resp.EmojiID == "" || resp.Emoji != "\U0001F44D\uFE0F" || resp.MaxReactions != 1 { + t.Fatalf("pre-handle response = %#v", resp) + } +} + func TestParseReactionTargetMessageID(t *testing.T) { messageID, err := parseReactionTargetMessageID(networkid.MessageID("616934195205767730")) if err != nil { @@ -565,12 +717,29 @@ func TestConvertMessageReactionsUsesEmbeddedHistory(t *testing.T) { if reactions[0].Emoji != "mxc://line/like" || reactions[0].Sender.Sender != "Uself" || !reactions[0].Sender.IsFromMe { t.Fatalf("predefined reaction = %#v", reactions[0]) } + if reactions[0].EmojiID != "predefined:2" { + t.Fatalf("predefined reaction EmojiID = %q", reactions[0].EmojiID) + } + predefinedMeta, ok := reactions[0].DBMetadata.(*ReactionMetadata) + if !ok || predefinedMeta.MatrixKey != reactions[0].Emoji || predefinedMeta.ReactionType.PredefinedReactionType != 2 { + t.Fatalf("predefined reaction metadata = %#v", reactions[0].DBMetadata) + } if want := time.UnixMilli(1784930400123); !reactions[0].Timestamp.Equal(want) { t.Fatalf("predefined reaction timestamp = %s, want %s", reactions[0].Timestamp, want) } if reactions[1].Emoji != "mxc://line/paid" || reactions[1].Sender.Sender != "Uother" || reactions[1].Sender.IsFromMe { t.Fatalf("paid reaction = %#v", reactions[1]) } + if reactions[1].EmojiID != "paid:paid-product:paid-emoji" { + t.Fatalf("paid reaction EmojiID = %q", reactions[1].EmojiID) + } + paidMeta, ok := reactions[1].DBMetadata.(*ReactionMetadata) + if !ok || paidMeta.MatrixKey != reactions[1].Emoji || + paidMeta.ReactionType.PaidReactionType == nil || + paidMeta.ReactionType.PaidReactionType.ProductID != paidType.ProductID || + paidMeta.ReactionType.PaidReactionType.EmojiID != paidType.EmojiID { + t.Fatalf("paid reaction metadata = %#v", reactions[1].DBMetadata) + } } func TestReactionUploadMXCRejectsEncryptedMedia(t *testing.T) { @@ -603,6 +772,45 @@ func TestQueueMessageReactionSyncSkipsSystemMarkers(t *testing.T) { } } +func TestLiveReactionSyncEventIsSenderAuthoritative(t *testing.T) { + lc := &LineClient{UserLogin: &bridgev2.UserLogin{ + UserLogin: &database.UserLogin{ID: "Uself"}, + }} + op := line.Operation{ + Param1: "616934195205767730", + CreatedTime: json.Number("1784930400123"), + } + reaction := &bridgev2.BackfillReaction{ + Sender: lc.eventSenderForMID("Uother"), + EmojiID: "paid:product:emoji", + Emoji: "mxc://line/custom", + } + + add := lc.liveReactionSyncEvent(op, "Cgroup", "Uother", reaction) + if add.Type != bridgev2.RemoteEventReactionSync || + add.PortalKey.ID != makePortalID("Cgroup") || + add.PortalKey.Receiver != "Uself" || + add.TargetMessage != "616934195205767730" { + t.Fatalf("add sync metadata = %#v", add) + } + if add.Reactions.HasAllUsers { + t.Fatal("single-sender live sync was marked authoritative for all users") + } + userSync := add.Reactions.Users["Uother"] + if userSync == nil || !userSync.HasAllReactions || len(userSync.Reactions) != 1 || userSync.Reactions[0] != reaction { + t.Fatalf("add user sync = %#v", userSync) + } + if want := time.UnixMilli(1784930400123); !add.Timestamp.Equal(want) { + t.Fatalf("add timestamp = %s, want %s", add.Timestamp, want) + } + + remove := lc.liveReactionSyncEvent(op, "Cgroup", "Uother", nil) + userSync = remove.Reactions.Users["Uother"] + if userSync == nil || !userSync.HasAllReactions || len(userSync.Reactions) != 0 { + t.Fatalf("remove user sync = %#v", userSync) + } +} + func TestResolveReactionSenderMID(t *testing.T) { lc := &LineClient{ UserLogin: &bridgev2.UserLogin{ diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index ba62364..1239219 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -2061,126 +2061,82 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { } func (lc *LineClient) handlePaidReaction(ctx context.Context, op line.Operation, param2 *line.ReactionPayload) { - prt := param2.Curr.PaidReactionType - mxc, err := lc.getPaidReactionMXC(ctx, prt) + ts, _ := op.CreatedTime.Int64() + reaction, err := lc.convertReaction( + ctx, + line.ReactionType{PaidReactionType: param2.Curr.PaidReactionType}, + lc.eventSenderForMID(op.Param3), + time.UnixMilli(ts), + ) if err != nil { lc.UserLogin.Bridge.Log.Error().Err(err).Msg("Failed to prepare paid reaction icon") return } - portalKey := networkid.PortalKey{ID: makePortalID(param2.ChatMid), Receiver: lc.UserLogin.ID} - - // A fresh add invalidates any prior remove-dedup entries for this - // message — otherwise a later removal would be silently skipped. - lc.clearReactionDedupEntries(op.Param1, true) - - ts, _ := op.CreatedTime.Int64() - lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, &simplevent.Reaction{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventReaction, - PortalKey: portalKey, - Timestamp: time.UnixMilli(ts), - Sender: lc.eventSenderForMID(op.Param3), - }, - TargetMessage: networkid.MessageID(op.Param1), - Emoji: mxc, - }) + lc.UserLogin.Bridge.QueueRemoteEvent( + lc.UserLogin, + lc.liveReactionSyncEvent(op, param2.ChatMid, reaction.Sender.Sender, reaction), + ) } func (lc *LineClient) handlePredefinedReaction(ctx context.Context, op line.Operation, chatMid string, prt int) { - if prt < 2 || prt > 7 { - lc.UserLogin.Bridge.Log.Error().Int("predefined_reaction_type", prt).Msg("Unknown predefined reaction type") - return - } - - portalKey := networkid.PortalKey{ID: makePortalID(chatMid), Receiver: lc.UserLogin.ID} - - mxc, err := lc.getPredefinedReactionMXC(ctx, prt) + ts, _ := op.CreatedTime.Int64() + reaction, err := lc.convertReaction( + ctx, + line.ReactionType{PredefinedReactionType: prt}, + lc.eventSenderForMID(op.Param3), + time.UnixMilli(ts), + ) if err != nil { - lc.UserLogin.Bridge.Log.Error().Err(err).Int("prt", prt).Msg("Failed to prepare predefined reaction icon") + lc.UserLogin.Bridge.Log.Error().Err(err).Int("predefined_reaction_type", prt).Msg("Failed to prepare predefined reaction icon") return } + lc.UserLogin.Bridge.QueueRemoteEvent( + lc.UserLogin, + lc.liveReactionSyncEvent(op, chatMid, reaction.Sender.Sender, reaction), + ) +} - dedupKey := op.Param1 + "\x00" + mxc - if _, loaded := lc.recentReactions.LoadOrStore(dedupKey, struct{}{}); loaded { - lc.UserLogin.Bridge.Log.Debug().Str("msg_id", op.Param1).Msg("Skipping duplicate predefined reaction") - return - } - - // A fresh add invalidates any prior remove-dedup entries for this - // message — otherwise a later removal of this (or a replacement) - // reaction would be silently skipped. - lc.clearReactionDedupEntries(op.Param1, true) - +func (lc *LineClient) liveReactionSyncEvent( + op line.Operation, + chatMid string, + sender networkid.UserID, + reaction *bridgev2.BackfillReaction, +) *simplevent.ReactionSync { ts, _ := op.CreatedTime.Int64() - lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, &simplevent.Reaction{ + reactions := []*bridgev2.BackfillReaction(nil) + if reaction != nil { + reactions = []*bridgev2.BackfillReaction{reaction} + } + return &simplevent.ReactionSync{ EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventReaction, - PortalKey: portalKey, + Type: bridgev2.RemoteEventReactionSync, + PortalKey: networkid.PortalKey{ID: makePortalID(chatMid), Receiver: lc.UserLogin.ID}, Timestamp: time.UnixMilli(ts), - Sender: lc.eventSenderForMID(op.Param3), }, TargetMessage: networkid.MessageID(op.Param1), - Emoji: mxc, - }) + Reactions: &bridgev2.ReactionSyncData{ + Users: map[networkid.UserID]*bridgev2.ReactionSyncUser{ + sender: { + Reactions: reactions, + HasAllReactions: true, + }, + }, + HasAllUsers: false, + }, + } } -// handleReactionRemove queues a RemoteEventReactionRemove for each candidate -// sender. Reactions are stored with EmojiID="" (see handlePaidReaction / -// handlePredefinedReaction), so the framework's reaction lookup finds the -// single row keyed by (target_message, sender) and redacts it. A miss is -// silently ignored by bridgev2, which lets callers safely queue multiple -// sender candidates when the previous reaction's actor is ambiguous. -// -// It also evicts stale add-dedup entries for the target message so that -// re-adding the same emoji after a clear isn't silently dropped by the -// recentReactions sync.Map. +// handleReactionRemove queues an authoritative empty reaction sync for each +// candidate sender. LINE only allows one reaction per sender, so this removes +// both legacy empty-ID rows and stable paid/predefined reaction IDs without +// needing the previous reaction type. func (lc *LineClient) handleReactionRemove(op line.Operation, chatMid string, senders []networkid.UserID) { - ts, _ := op.CreatedTime.Int64() - portalKey := networkid.PortalKey{ID: makePortalID(chatMid), Receiver: lc.UserLogin.ID} - for _, sender := range senders { - dedupKey := op.Param1 + "\x00remove\x00" + string(sender) - if _, loaded := lc.recentReactions.LoadOrStore(dedupKey, struct{}{}); loaded { - lc.UserLogin.Bridge.Log.Debug().Str("msg_id", op.Param1).Str("sender", string(sender)).Msg("Skipping duplicate reaction removal") - continue - } - lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, &simplevent.Reaction{ - EventMeta: simplevent.EventMeta{ - Type: bridgev2.RemoteEventReactionRemove, - PortalKey: portalKey, - Timestamp: time.UnixMilli(ts), - Sender: lc.eventSenderForMID(string(sender)), - }, - TargetMessage: networkid.MessageID(op.Param1), - }) + lc.UserLogin.Bridge.QueueRemoteEvent( + lc.UserLogin, + lc.liveReactionSyncEvent(op, chatMid, sender, nil), + ) } - - lc.clearReactionDedupEntries(op.Param1, false) -} - -// clearReactionDedupEntries evicts recentReactions entries for the given -// message. The recentReactions sync.Map dedups concurrent 139/140 events -// from LINE; without periodic cleanup, the keys accumulate and silently -// block legitimate later events (e.g. add → remove → add of the same -// emoji, or remove → add → remove sequences). We use the inverse-direction -// event as the cleanup trigger: an add clears stale remove-dedup entries -// (removeOnly=true), a remove clears stale add-dedup entries -// (removeOnly=false). -func (lc *LineClient) clearReactionDedupEntries(msgID string, removeOnly bool) { - prefix := msgID + "\x00" - lc.recentReactions.Range(func(k, _ any) bool { - ks, ok := k.(string) - if !ok { - return true - } - if !strings.HasPrefix(ks, prefix) { - return true - } - if strings.Contains(ks, "\x00remove\x00") == removeOnly { - lc.recentReactions.Delete(ks) - } - return true - }) } func (lc *LineClient) syncSingleChat(ctx context.Context, op line.Operation) { From ef2373e0bd73a54d3940fc1dd9a143f0cc2464b8 Mon Sep 17 00:00:00 2001 From: highesttt Date: Tue, 28 Jul 2026 14:14:51 -0400 Subject: [PATCH 2/4] fix: skip invalid stored reactions --- pkg/connector/reaction.go | 2 +- pkg/connector/reaction_test.go | 15 +++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/pkg/connector/reaction.go b/pkg/connector/reaction.go index 64661aa..6e210a8 100644 --- a/pkg/connector/reaction.go +++ b/pkg/connector/reaction.go @@ -548,7 +548,7 @@ func storedLineReactionForMatrixKey(key string, reactions []*database.Reaction) } ref, err := newLineReactionRef(meta.ReactionType) if err != nil || (reaction.EmojiID != "" && reaction.EmojiID != ref.networkEmojiID()) { - return lineReactionRef{}, false + continue } if hasFound && !found.equal(ref) { return lineReactionRef{}, false diff --git a/pkg/connector/reaction_test.go b/pkg/connector/reaction_test.go index 0ec1c84..5bb86ea 100644 --- a/pkg/connector/reaction_test.go +++ b/pkg/connector/reaction_test.go @@ -536,6 +536,21 @@ func TestStoredLineReactionForMatrixKey(t *testing.T) { if !ok || ref.networkEmojiID() != "paid:product:emoji" { t.Fatalf("stored reaction = %#v, %v", ref, ok) } + malformed := &database.Reaction{ + EmojiID: "predefined:999", + Metadata: &ReactionMetadata{ + MatrixKey: key, + ReactionType: line.ReactionType{PredefinedReactionType: 999}, + }, + } + mismatched := &database.Reaction{ + EmojiID: "paid:different:id", + Metadata: valid.Metadata, + } + ref, ok = storedLineReactionForMatrixKey(key, []*database.Reaction{malformed, mismatched, valid}) + if !ok || ref.networkEmojiID() != "paid:product:emoji" { + t.Fatalf("stored reaction after malformed rows = %#v, %v", ref, ok) + } if _, ok = storedLineReactionForMatrixKey("mxc://line/arbitrary", []*database.Reaction{valid}); ok { t.Fatal("arbitrary MXC was accepted") } From fc15510975d2151358ab813bc0255f4ba4750310 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 30 Jul 2026 10:50:49 -0400 Subject: [PATCH 3/4] fix: race condition stale tokens/in flight requests --- pkg/connector/auth_recovery.go | 94 ++++++- pkg/connector/auth_recovery_test.go | 216 +++++++++++++-- pkg/connector/client.go | 46 ++-- pkg/connector/creategroup.go | 28 +- pkg/connector/e2ee_keys.go | 32 +-- pkg/connector/handle_message.go | 14 +- pkg/connector/handlers/audio.go | 3 +- pkg/connector/handlers/file.go | 3 +- pkg/connector/handlers/handler.go | 41 ++- pkg/connector/handlers/handler_test.go | 75 ++++-- pkg/connector/handlers/image.go | 3 +- pkg/connector/handlers/post_notification.go | 6 +- .../handlers/post_notification_test.go | 18 -- pkg/connector/handlers/video.go | 3 +- pkg/connector/sync.go | 245 +++++++----------- pkg/connector/sync_test.go | 35 ++- pkg/connector/userinfo.go | 74 ++---- 17 files changed, 542 insertions(+), 394 deletions(-) diff --git a/pkg/connector/auth_recovery.go b/pkg/connector/auth_recovery.go index 853d764..3ee5901 100644 --- a/pkg/connector/auth_recovery.go +++ b/pkg/connector/auth_recovery.go @@ -9,7 +9,7 @@ import ( type lineCallDeps[T any] struct { newClient func() *line.Client - recover func(context.Context) error + recover func(context.Context, *line.Client, error) (*line.Client, error) isAuthError func(error) bool call func(*line.Client) (T, error) } @@ -27,13 +27,23 @@ func callLineWithRecovery[T any](ctx context.Context, client *line.Client, deps return client, res, err } - if errRecover := deps.recover(ctx); errRecover != nil { + recoveredClient, errRecover := deps.recover(ctx, client, err) + if errRecover != nil { var zero T return client, zero, fmt.Errorf("failed to recover token after LINE auth error (%w): %w", err, errRecover) } + if recoveredClient == nil { + return client, res, err + } - client = deps.newClient() + client = recoveredClient res, err = deps.call(client) + if line.IsLoggedOut(err) { + // The retry is the final attempt, but a current-token logout still needs + // to transition the login to BAD_CREDENTIALS. The source-aware recovery + // callback will ignore it if another token rotation made this retry stale. + _, _ = deps.recover(ctx, client, err) + } return client, res, err } @@ -44,10 +54,74 @@ func (lc *LineClient) isTokenError(err error) bool { if lc.isSessionInvalidated() { return false } + return line.IsAuthError(err) +} + +// recoverClientAfterAuthError classifies an auth error using the exact client +// that produced it. Logged-out responses from an older access token are safe to +// retry after a concurrent refresh/re-login; the same response from the current +// token is a genuine forced logout and must invalidate the session. +func (lc *LineClient) recoverClientAfterAuthError(ctx context.Context, failedClient *line.Client, err error) (*line.Client, error) { + if !lc.isTokenError(err) { + return nil, nil + } + + // Wait behind any in-flight refresh/re-login before comparing tokens. This + // makes the comparison authoritative even when the failed request completed + // while another goroutine was rotating the access token. + lc.recoverMu.Lock() + if ctx.Err() != nil { + lc.recoverMu.Unlock() + return nil, ctx.Err() + } + + currentToken := lc.getAccessToken() + if failedClient != nil && failedClient.AccessToken != "" && currentToken != "" && failedClient.AccessToken != currentToken && !lc.isSessionInvalidated() { + if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { + lc.UserLogin.Bridge.Log.Debug(). + Bool("logged_out", line.IsLoggedOut(err)). + Bool("stale_access_token", true). + Msg("Retrying LINE request after response from stale access token") + } + lc.recoverMu.Unlock() + return newLineAPIClient(currentToken), nil + } + if line.IsLoggedOut(err) { - return false + lc.markLoggedOutByOtherClientLocked(ctx, err) + lc.recoverMu.Unlock() + return nil, nil } - return line.IsAuthError(err) + if lc.recoveryStopped || lc.superseded.Load() { + lc.recoverMu.Unlock() + return nil, errLineClientSuperseded + } + if lc.isSessionInvalidated() { + lc.recoverMu.Unlock() + return nil, errLineSessionInvalidated + } + lc.recoverMu.Unlock() + + recoveryToken := lc.getAccessToken() + if errRecover := recoverLineToken(lc, ctx); errRecover != nil { + if line.IsLoggedOut(errRecover) { + // Refresh/re-login errors come from the token that was current when + // recovery started. Classify them with the same source-aware path in + // case another serialized recovery rotated that token first. + return lc.recoverClientAfterAuthError(ctx, newLineAPIClient(recoveryToken), errRecover) + } + return nil, errRecover + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + if lc.superseded.Load() { + return nil, errLineClientSuperseded + } + if lc.isSessionInvalidated() { + return nil, errLineSessionInvalidated + } + return lc.newClient(), nil } func (lc *LineClient) callLine(ctx context.Context, call func(*line.Client) error) (*line.Client, error) { @@ -57,15 +131,12 @@ func (lc *LineClient) callLine(ctx context.Context, call func(*line.Client) erro func (lc *LineClient) callLineUsing(ctx context.Context, client *line.Client, call func(*line.Client) error) (*line.Client, error) { client, _, err := callLineWithRecovery(ctx, client, lineCallDeps[struct{}]{ newClient: func() *line.Client { return lc.newClient() }, - recover: func(ctx context.Context) error { return recoverLineToken(lc, ctx) }, + recover: lc.recoverClientAfterAuthError, isAuthError: lc.isTokenError, call: func(client *line.Client) (struct{}, error) { return struct{}{}, call(client) }, }) - if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) - } return client, err } @@ -76,12 +147,9 @@ func callLineResult[T any](lc *LineClient, ctx context.Context, call func(*line. func callLineResultUsing[T any](lc *LineClient, ctx context.Context, client *line.Client, call func(*line.Client) (T, error)) (*line.Client, T, error) { client, res, err := callLineWithRecovery(ctx, client, lineCallDeps[T]{ newClient: func() *line.Client { return lc.newClient() }, - recover: func(ctx context.Context) error { return recoverLineToken(lc, ctx) }, + recover: lc.recoverClientAfterAuthError, isAuthError: lc.isTokenError, call: call, }) - if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) - } return client, res, err } diff --git a/pkg/connector/auth_recovery_test.go b/pkg/connector/auth_recovery_test.go index e6f2453..1eed26c 100644 --- a/pkg/connector/auth_recovery_test.go +++ b/pkg/connector/auth_recovery_test.go @@ -90,9 +90,12 @@ func TestCallLineWithRecovery(t *testing.T) { newClient: func() *line.Client { return line.NewClient("token") }, - recover: func(context.Context) error { + recover: func(context.Context, *line.Client, error) (*line.Client, error) { recoveries++ - return tt.recoverErr + if tt.recoverErr != nil { + return nil, tt.recoverErr + } + return line.NewClient("recovered"), nil }, isAuthError: line.IsAuthError, call: func(*line.Client) (struct{}, error) { @@ -138,8 +141,8 @@ func TestCallLineWithRecoveryReusesClientUntilRecovery(t *testing.T) { newClients++ return refreshedClient }, - recover: func(context.Context) error { - return nil + recover: func(context.Context, *line.Client, error) (*line.Client, error) { + return refreshedClient, nil }, isAuthError: line.IsAuthError, call: func(client *line.Client) (struct{}, error) { @@ -156,8 +159,8 @@ func TestCallLineWithRecoveryReusesClientUntilRecovery(t *testing.T) { if client != refreshedClient { t.Fatal("expected recovered client to be returned") } - if newClients != 1 { - t.Fatalf("new clients = %d, want 1", newClients) + if newClients != 0 { + t.Fatalf("new clients = %d, want 0 because recovery returned the retry client", newClients) } if len(calls) != 2 || calls[0] != "initial" || calls[1] != "refreshed" { t.Fatalf("calls used clients %v, want [initial refreshed]", calls) @@ -174,7 +177,9 @@ func TestCallLineWithRecoveryUsesProvidedClientWithoutRecreating(t *testing.T) { newClients++ return line.NewClient("unexpected") }, - recover: func(context.Context) error { return nil }, + recover: func(context.Context, *line.Client, error) (*line.Client, error) { + return line.NewClient("unexpected"), nil + }, isAuthError: line.IsAuthError, call: func(client *line.Client) (struct{}, error) { if client.AccessToken != "initial" { @@ -194,16 +199,16 @@ func TestCallLineWithRecoveryUsesProvidedClientWithoutRecreating(t *testing.T) { } } -func TestLineClientIsTokenErrorExcludesNonRecoverableErrors(t *testing.T) { +func TestLineClientIsTokenErrorClassifiesRecoverableErrors(t *testing.T) { lc := &LineClient{} if !lc.isTokenError(errAuthRequired) { t.Fatal("expected auth-required error to be classified as token error") } - if lc.isTokenError(errLoggedOut) { - t.Fatal("logged-out sessions must not trigger token recovery") + if !lc.isTokenError(errLoggedOut) { + t.Fatal("logged-out sessions must reach source-aware auth handling") } - if lc.isTokenError(errSenderKey) { - t.Fatal("invalid sender key sessions must not trigger token recovery") + if !lc.isTokenError(errSenderKey) { + t.Fatal("invalid sender key sessions must reach source-aware auth handling") } lc.sessionInvalidated = true if lc.isTokenError(errAuthRequired) { @@ -218,6 +223,125 @@ func TestLineClientIsTokenErrorExcludesNonRecoverableErrors(t *testing.T) { } } +func TestCallLineUsingRetriesStaleLogoutWithCurrentToken(t *testing.T) { + lc := &LineClient{AccessToken: "current-token"} + var calls []string + client, err := lc.callLineUsing(context.Background(), line.NewClient("old-token"), func(client *line.Client) error { + calls = append(calls, client.AccessToken) + if client.AccessToken == "old-token" { + return errLoggedOut + } + return nil + }) + if err != nil { + t.Fatalf("callLineUsing returned error: %v", err) + } + if client == nil || client.AccessToken != "current-token" { + t.Fatalf("client = %#v, want current-token", client) + } + if len(calls) != 2 || calls[0] != "old-token" || calls[1] != "current-token" { + t.Fatalf("call tokens = %v, want [old-token current-token]", calls) + } + if lc.isSessionInvalidated() { + t.Fatal("stale logout invalidated the current session") + } +} + +func TestRecoveryLoggedOutErrorInvalidatesCurrentSession(t *testing.T) { + oldRecover := recoverLineToken + t.Cleanup(func() { + recoverLineToken = oldRecover + }) + recoverLineToken = func(*LineClient, context.Context) error { + return errLoggedOut + } + + lc := &LineClient{AccessToken: "current-token"} + retryClient, err := lc.recoverClientAfterAuthError(context.Background(), line.NewClient("current-token"), errAuthRequired) + if err != nil { + t.Fatalf("recoverClientAfterAuthError returned error: %v", err) + } + if retryClient != nil { + t.Fatalf("retry client = %#v, want nil", retryClient) + } + if lc.hasAccessToken() || !lc.isSessionInvalidated() { + t.Fatal("logged-out recovery error did not invalidate the current session") + } +} + +func TestConcurrentOldTokenFailuresWaitForSingleRecovery(t *testing.T) { + oldRecover := recoverLineToken + t.Cleanup(func() { + recoverLineToken = oldRecover + }) + + lc := &LineClient{AccessToken: "old-token"} + recoveryStarted := make(chan struct{}) + allowRecovery := make(chan struct{}) + var recoveryCalls atomic.Int32 + recoverLineToken = func(lc *LineClient, ctx context.Context) error { + return lc.runTokenRecovery(ctx, func(context.Context) error { + if recoveryCalls.Add(1) == 1 { + close(recoveryStarted) + } + <-allowRecovery + lc.setTokens("new-token", "") + return nil + }) + } + + primaryDone := make(chan error, 1) + go func() { + _, err := lc.callLineUsing(context.Background(), line.NewClient("old-token"), func(client *line.Client) error { + if client.AccessToken == "old-token" { + return errAuthRequired + } + return nil + }) + primaryDone <- err + }() + + select { + case <-recoveryStarted: + case <-time.After(time.Second): + t.Fatal("primary recovery did not start") + } + + const staleCalls = 8 + var started sync.WaitGroup + started.Add(staleCalls) + staleDone := make(chan error, staleCalls) + for range staleCalls { + go func() { + _, err := lc.callLineUsing(context.Background(), line.NewClient("old-token"), func(client *line.Client) error { + if client.AccessToken == "old-token" { + started.Done() + return errLoggedOut + } + return nil + }) + staleDone <- err + }() + } + started.Wait() + close(allowRecovery) + + if err := <-primaryDone; err != nil { + t.Fatalf("primary call returned error: %v", err) + } + for range staleCalls { + if err := <-staleDone; err != nil { + t.Fatalf("stale call returned error: %v", err) + } + } + if got := recoveryCalls.Load(); got != 1 { + t.Fatalf("recovery calls = %d, want 1", got) + } + if lc.getAccessToken() != "new-token" || lc.isSessionInvalidated() { + t.Fatal("concurrent stale failures clobbered the recovered session") + } +} + func TestRunTokenRecoverySkipsRecentRecovery(t *testing.T) { lc := &LineClient{recoverTime: time.Now()} var calls int @@ -308,8 +432,15 @@ func TestRecoverTokenDoesNotReloginAfterCancellation(t *testing.T) { } } -func TestForcedLogoutWinsOverInFlightRecovery(t *testing.T) { +func TestStaleForcedLogoutDoesNotClobberInFlightRecovery(t *testing.T) { lc := &LineClient{AccessToken: "old-token"} + runCtx, _, started := lc.beginRun(context.Background()) + if !started { + t.Fatal("beginRun unexpectedly rejected startup") + } + defer lc.wg.Done() + defer lc.cancelActiveRun() + recoveryStarted := make(chan struct{}) allowRecovery := make(chan struct{}) recoveryDone := make(chan error, 1) @@ -323,10 +454,14 @@ func TestForcedLogoutWinsOverInFlightRecovery(t *testing.T) { }() <-recoveryStarted - logoutDone := make(chan struct{}) + type recoveryResult struct { + client *line.Client + err error + } + logoutDone := make(chan recoveryResult, 1) go func() { - lc.markLoggedOutByOtherClient(context.Background(), errLoggedOut) - close(logoutDone) + client, err := lc.recoverClientAfterAuthError(context.Background(), line.NewClient("old-token"), errLoggedOut) + logoutDone <- recoveryResult{client: client, err: err} }() close(allowRecovery) @@ -334,15 +469,52 @@ func TestForcedLogoutWinsOverInFlightRecovery(t *testing.T) { t.Fatalf("recovery returned error: %v", err) } select { - case <-logoutDone: + case result := <-logoutDone: + if result.err != nil { + t.Fatalf("stale logout handling returned error: %v", result.err) + } + if result.client == nil || result.client.AccessToken != "recovered-token" { + t.Fatalf("retry client = %#v, want recovered-token", result.client) + } case <-time.After(time.Second): - t.Fatal("forced logout did not complete after recovery") + t.Fatal("stale logout handling did not complete after recovery") } - if lc.hasAccessToken() { - t.Fatal("in-flight recovery resurrected the invalidated session") + if lc.getAccessToken() != "recovered-token" { + t.Fatalf("access token = %q, want recovered-token", lc.getAccessToken()) } - if !lc.isSessionInvalidated() { - t.Fatal("session was not invalidated after in-flight recovery") + if lc.isSessionInvalidated() { + t.Fatal("stale logout invalidated the recovered session") + } + select { + case <-runCtx.Done(): + t.Fatal("stale logout canceled the active run") + default: + } +} + +func TestCurrentTokenLoggedOutErrorsInvalidateSession(t *testing.T) { + tests := []struct { + name string + err error + }{ + {name: "client logged out", err: errLoggedOut}, + {name: "invalid sender key", err: errSenderKey}, + {name: "request need login", err: errors.New(`SSE error: 401: {"code":10004,"message":"REQUEST_NEED_LOGIN"}`)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + lc := &LineClient{AccessToken: "current-token"} + retryClient, err := lc.recoverClientAfterAuthError(context.Background(), line.NewClient("current-token"), tt.err) + if err != nil { + t.Fatalf("recoverClientAfterAuthError returned error: %v", err) + } + if retryClient != nil { + t.Fatalf("retry client = %#v, want nil", retryClient) + } + if lc.hasAccessToken() || !lc.isSessionInvalidated() { + t.Fatal("current-token logout did not invalidate the session") + } + }) } } diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 4415b17..72eda47 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -315,27 +315,16 @@ func (lc *LineClient) isLoggedOut(err error) bool { return line.IsLoggedOut(err) } -func (lc *LineClient) shouldAttemptTokenRecovery(ctx context.Context, err error) bool { - if err == nil { - return false - } - if ctx.Err() != nil || lc.superseded.Load() || lc.isSessionInvalidated() { - return false - } - if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) - return false - } - return lc.isRefreshRequired(err) || line.IsUnauthorizedStatus(err) -} - func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) { - // Serialize invalidation with token recovery. If a refresh/re-login was - // already in flight, it may finish first, but this transition always runs - // afterward so recovery cannot resurrect a forcefully logged-out session. lc.recoverMu.Lock() defer lc.recoverMu.Unlock() + lc.markLoggedOutByOtherClientLocked(ctx, err) +} +// markLoggedOutByOtherClientLocked applies the persistent forced-logout +// transition. The caller must hold recoverMu so token rotation and invalidation +// cannot interleave. +func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err error) { if lc.UserLogin == nil { lc.invalidateAccessToken() line.InvalidateOBSTokenCache() @@ -783,15 +772,20 @@ func (lc *LineClient) refreshLoginE2EEKeys(res *line.LoginResult, meta *UserLogi } func (lc *LineClient) ensureValidToken(ctx context.Context) error { - return lc.ensureValidTokenWith( - ctx, - func(ctx context.Context) error { - _, err := getProfileWithToken(ctx, lc.getAccessToken()) - return err - }, - lc.refreshAndSave, - lc.tryLogin, - ) + _, _, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Profile, error) { + return getProfileWithToken(ctx, client.AccessToken) + }) + if err == nil { + return nil + } + if ctx.Err() != nil { + return ctx.Err() + } + if line.IsAuthError(err) { + return err + } + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("GetProfile failed with non-auth error, continuing anyway") + return nil } func (lc *LineClient) ensureValidTokenWith( diff --git a/pkg/connector/creategroup.go b/pkg/connector/creategroup.go index 73b6d2d..28f9646 100644 --- a/pkg/connector/creategroup.go +++ b/pkg/connector/creategroup.go @@ -32,18 +32,11 @@ func (lc *LineClient) CreateGroup(ctx context.Context, params *bridgev2.GroupCre name = params.Name.Name } - client := lc.newClient() - var chat *line.Chat - var err error chatType := 1 // ROOM: members join automatically. lineName := name - chat, err = client.CreateChat(participantMids, lineName, chatType) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - chat, err = client.CreateChat(participantMids, lineName, chatType) - } - } + _, chat, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Chat, error) { + return client.CreateChat(participantMids, lineName, chatType) + }) if err != nil { return nil, fmt.Errorf("failed to create LINE chat: %w", err) } @@ -274,16 +267,11 @@ func (lc *LineClient) registerGroupKey(ctx context.Context, chatMid string, memb keyIds = append(keyIds, selfRawID) encryptedKeys = append(encryptedKeys, selfEncryptedKey) - if err := client.RegisterE2EEGroupKey(1, chatMid, apiMembers, keyIds, encryptedKeys); err != nil { - if lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - err = client.RegisterE2EEGroupKey(1, chatMid, apiMembers, keyIds, encryptedKeys) - } - } - if err != nil { - return fmt.Errorf("registerE2EEGroupKey failed: %w", err) - } + _, err = lc.callLineUsing(ctx, client, func(client *line.Client) error { + return client.RegisterE2EEGroupKey(1, chatMid, apiMembers, keyIds, encryptedKeys) + }) + if err != nil { + return fmt.Errorf("registerE2EEGroupKey failed: %w", err) } lc.UserLogin.Bridge.Log.Info(). diff --git a/pkg/connector/e2ee_keys.go b/pkg/connector/e2ee_keys.go index 51b2f0b..2fc2821 100644 --- a/pkg/connector/e2ee_keys.go +++ b/pkg/connector/e2ee_keys.go @@ -39,7 +39,7 @@ func (lc *LineClient) fetchAndUnwrapGroupKey(ctx context.Context, chatMid string } client := lc.newClient() - fetch := func() (*line.E2EEGroupSharedKey, error) { + fetch := func(client *line.Client) (*line.E2EEGroupSharedKey, error) { var sharedKey *line.E2EEGroupSharedKey var err error if groupKeyID > 0 { @@ -50,7 +50,7 @@ func (lc *LineClient) fetchAndUnwrapGroupKey(ctx context.Context, chatMid string return sharedKey, groupKeyFetchError(groupKeyID, err) } - sharedKey, err := fetch() + client, sharedKey, err := callLineResultUsing(lc, ctx, client, fetch) // No group key exists yet — auto-register one so the group can use E2EE. if err != nil && line.IsGroupKeyNotFound(err) { lc.UserLogin.Bridge.Log.Info().Str("chat_mid", chatMid). @@ -60,16 +60,7 @@ func (lc *LineClient) fetchAndUnwrapGroupKey(ctx context.Context, chatMid string Msg("Auto-register group key failed") return fmt.Errorf("auto-register group key: %w", registerErr) } - sharedKey, err = fetch() - } - // Token recovery for other error types - if err != nil && !line.IsNoUsableE2EEGroupKey(err) && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - sharedKey, err = fetch() - } else { - return fmt.Errorf("failed to recover token before fetching group key: %w", errRecover) - } + client, sharedKey, err = callLineResultUsing(lc, ctx, client, fetch) } if err != nil { return err @@ -115,7 +106,7 @@ func (lc *LineClient) fetchAndUnwrapGroupKey(ctx context.Context, chatMid string if registerErr := lc.autoRegisterGroupKey(ctx, chatMid); registerErr != nil { return fmt.Errorf("%w (fresh group key registration failed: %v)", err, registerErr) } - sharedKey, err = fetch() + _, sharedKey, err = callLineResultUsing(lc, ctx, client, fetch) if err != nil { return fmt.Errorf("failed to fetch fresh group key after registration: %w", err) } @@ -242,18 +233,11 @@ func joinedGroupMemberMIDs(group *line.GroupExtra, ownMID string) ([]string, boo // Pending invitees are deliberately excluded: LINE validates group keys against the // current joined-member set and rejects keys that include invitees. func (lc *LineClient) getChatMemberMIDs(ctx context.Context, chatMid string) ([]string, bool, error) { - client := lc.newClient() - chats, err := client.GetChats([]string{chatMid}, true, true) + _, chats, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetChatsResponse, error) { + return client.GetChats([]string{chatMid}, true, true) + }) if err != nil { - if lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - chats, err = client.GetChats([]string{chatMid}, true, true) - } - } - if err != nil { - return nil, false, fmt.Errorf("getChats failed for %s: %w", chatMid, err) - } + return nil, false, fmt.Errorf("getChats failed for %s: %w", chatMid, err) } if len(chats.Chats) == 0 { return nil, false, fmt.Errorf("chat %s not found", chatMid) diff --git a/pkg/connector/handle_message.go b/pkg/connector/handle_message.go index 685a050..3e92a56 100644 --- a/pkg/connector/handle_message.go +++ b/pkg/connector/handle_message.go @@ -29,15 +29,11 @@ const ( func (lc *LineClient) newMessageHandler() *handlers.Handler { return &handlers.Handler{ - Log: lc.UserLogin.Bridge.Log, - HTTPClient: lc.HTTPClient, - RecoverToken: lc.recoverToken, - ShouldRecover: lc.shouldAttemptTokenRecovery, - IsRefreshRequired: lc.isRefreshRequired, - IsLoggedOut: lc.isLoggedOut, - HandleLoggedOut: lc.markLoggedOutByOtherClient, - NewClient: func() *line.Client { return lc.newClient() }, - DecryptMedia: lc.decryptImageData, + Log: lc.UserLogin.Bridge.Log, + HTTPClient: lc.HTTPClient, + RecoverClient: lc.recoverClientAfterAuthError, + NewClient: func() *line.Client { return lc.newClient() }, + DecryptMedia: lc.decryptImageData, } } diff --git a/pkg/connector/handlers/audio.go b/pkg/connector/handlers/audio.go index 54e4832..1e2236b 100644 --- a/pkg/connector/handlers/audio.go +++ b/pkg/connector/handlers/audio.go @@ -48,10 +48,11 @@ func (h *Handler) ConvertAudio(ctx context.Context, portal *bridgev2.Portal, int talkMetaMessageID := obsTalkMetaMessageID(data.ID, isPlainMedia) audioData, err := client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, sid, downloadOptions) - if newClient, ok := h.tryRecoverClient(ctx, err); ok { + if newClient, ok := h.tryRecoverClient(ctx, client, err); ok { client = newClient audioData, err = client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, sid, downloadOptions) } + h.handleFinalAuthError(ctx, client, err) if err != nil { h.Log.Warn(). diff --git a/pkg/connector/handlers/file.go b/pkg/connector/handlers/file.go index d4be6f1..e2eac47 100644 --- a/pkg/connector/handlers/file.go +++ b/pkg/connector/handlers/file.go @@ -39,10 +39,11 @@ func (h *Handler) ConvertFile(ctx context.Context, portal *bridgev2.Portal, inte talkMetaMessageID := obsTalkMetaMessageID(data.ID, isPlainMedia) fileData, err := client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, sid, downloadOptions) - if newClient, ok := h.tryRecoverClient(ctx, err); ok { + if newClient, ok := h.tryRecoverClient(ctx, client, err); ok { client = newClient fileData, err = client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, sid, downloadOptions) } + h.handleFinalAuthError(ctx, client, err) if err != nil { h.Log.Warn(). diff --git a/pkg/connector/handlers/handler.go b/pkg/connector/handlers/handler.go index 3a2b1cd..1e34086 100644 --- a/pkg/connector/handlers/handler.go +++ b/pkg/connector/handlers/handler.go @@ -19,12 +19,9 @@ type Handler struct { Log zerolog.Logger HTTPClient *http.Client - // RecoverToken attempts to restore a valid session by refreshing or re-logging in. - RecoverToken func(ctx context.Context) error - ShouldRecover func(ctx context.Context, err error) bool - IsRefreshRequired func(err error) bool - IsLoggedOut func(err error) bool - HandleLoggedOut func(ctx context.Context, err error) + // RecoverClient classifies auth errors using the client that made the failed + // request, and returns a client that may be used for one retry. + RecoverClient func(ctx context.Context, failedClient *line.Client, err error) (*line.Client, error) // NewClient creates a new LINE API client with the current access token. NewClient func() *line.Client @@ -77,26 +74,26 @@ func mediaDownloadFailure(kind string, err error, relatesTo *event.RelatesTo) (* // tryRecoverClient attempts token recovery on auth errors and returns a fresh client. // Returns (newClient, true) on success, (nil, false) if recovery was not needed or failed. -func (h *Handler) tryRecoverClient(ctx context.Context, err error) (*line.Client, bool) { - if err == nil { +func (h *Handler) tryRecoverClient(ctx context.Context, failedClient *line.Client, err error) (*line.Client, bool) { + if err == nil || h.RecoverClient == nil { return nil, false } - if h.IsLoggedOut(err) { - if h.HandleLoggedOut != nil { - h.HandleLoggedOut(ctx, err) - } + recoveredClient, errRecover := h.RecoverClient(ctx, failedClient, err) + if errRecover != nil { + h.Log.Warn().Err(errRecover).Msg("Failed to recover token for media download") return nil, false } - if h.ShouldRecover != nil { - if !h.ShouldRecover(ctx, err) { - return nil, false - } - } else if !line.IsUnauthorizedStatus(err) && !h.IsRefreshRequired(err) { - return nil, false + return recoveredClient, recoveredClient != nil +} + +// handleFinalAuthError applies forced-logout handling to the one allowed retry +// without requesting another retry. Source-aware recovery will ignore the error +// if another token rotation already made the retry client stale. +func (h *Handler) handleFinalAuthError(ctx context.Context, failedClient *line.Client, err error) { + if !line.IsLoggedOut(err) || h.RecoverClient == nil { + return } - if errRecover := h.RecoverToken(ctx); errRecover != nil { - h.Log.Warn().Err(errRecover).Msg("Failed to recover token for media download") - return nil, false + if _, errRecover := h.RecoverClient(ctx, failedClient, err); errRecover != nil { + h.Log.Warn().Err(errRecover).Msg("Failed to handle LINE logout after media retry") } - return h.NewClient(), true } diff --git a/pkg/connector/handlers/handler_test.go b/pkg/connector/handlers/handler_test.go index 606adad..4e16257 100644 --- a/pkg/connector/handlers/handler_test.go +++ b/pkg/connector/handlers/handler_test.go @@ -11,60 +11,79 @@ import ( "github.com/highesttt/matrix-line-messenger/pkg/line" ) -func TestTryRecoverClientUsesShouldRecover(t *testing.T) { +func TestTryRecoverClientPassesOriginatingClient(t *testing.T) { errAuth := errors.New("SSE error: 401") var recoverCalled bool + failedClient := line.NewClient("failed-token") h := &Handler{ - ShouldRecover: func(context.Context, error) bool { - return false - }, - IsLoggedOut: func(error) bool { - return false - }, - IsRefreshRequired: func(error) bool { - return true - }, - RecoverToken: func(context.Context) error { + RecoverClient: func(_ context.Context, client *line.Client, err error) (*line.Client, error) { recoverCalled = true - return nil + if client != failedClient { + t.Fatalf("failed client = %#v, want originating client", client) + } + if !errors.Is(err, errAuth) { + t.Fatalf("auth error = %v, want %v", err, errAuth) + } + return nil, nil }, } - client, ok := h.tryRecoverClient(context.Background(), errAuth) + client, ok := h.tryRecoverClient(context.Background(), failedClient, errAuth) if ok || client != nil { t.Fatalf("tryRecoverClient returned client=%v ok=%v, want no recovery", client, ok) } - if recoverCalled { - t.Fatal("RecoverToken was called despite ShouldRecover returning false") + if !recoverCalled { + t.Fatal("RecoverClient was not called") } } func TestTryRecoverClientRecoversOBSObjectInfoUnauthorized(t *testing.T) { recoveredClient := line.NewClient("refreshed-token") + failedClient := line.NewClient("expired-token") var recoverCalled bool h := &Handler{ - ShouldRecover: func(_ context.Context, err error) bool { - return line.IsUnauthorizedStatus(err) - }, - IsLoggedOut: func(error) bool { - return false - }, - RecoverToken: func(context.Context) error { + RecoverClient: func(_ context.Context, client *line.Client, err error) (*line.Client, error) { recoverCalled = true - return nil - }, - NewClient: func() *line.Client { - return recoveredClient + if client != failedClient { + t.Fatalf("failed client = %#v, want originating client", client) + } + if !line.IsUnauthorizedStatus(err) { + t.Fatalf("error = %v, want unauthorized status", err) + } + return recoveredClient, nil }, } - client, ok := h.tryRecoverClient(context.Background(), errors.New("OBS object info failed (401): unauthorized")) + client, ok := h.tryRecoverClient(context.Background(), failedClient, errors.New("OBS object info failed (401): unauthorized")) if !ok || client != recoveredClient { t.Fatalf("tryRecoverClient returned client=%v ok=%v, want refreshed client", client, ok) } if !recoverCalled { - t.Fatal("RecoverToken was not called for OBS object-info 401") + t.Fatal("RecoverClient was not called for OBS object-info 401") + } +} + +func TestHandleFinalAuthErrorKeepsRetrySourceClient(t *testing.T) { + errLoggedOut := errors.New("V3_TOKEN_CLIENT_LOGGED_OUT") + retryClient := line.NewClient("retry-token") + var calls int + h := &Handler{ + RecoverClient: func(_ context.Context, client *line.Client, err error) (*line.Client, error) { + calls++ + if client != retryClient { + t.Fatalf("failed client = %#v, want retry client", client) + } + if !errors.Is(err, errLoggedOut) { + t.Fatalf("error = %v, want logged-out error", err) + } + return nil, nil + }, + } + + h.handleFinalAuthError(context.Background(), retryClient, errLoggedOut) + if calls != 1 { + t.Fatalf("RecoverClient calls = %d, want 1", calls) } } diff --git a/pkg/connector/handlers/image.go b/pkg/connector/handlers/image.go index 1201d9c..83f315b 100644 --- a/pkg/connector/handlers/image.go +++ b/pkg/connector/handlers/image.go @@ -50,7 +50,7 @@ func (h *Handler) ConvertImage(ctx context.Context, portal *bridgev2.Portal, int } // Refresh token if we get a 401 - if newClient, ok := h.tryRecoverClient(ctx, err); ok { + if newClient, ok := h.tryRecoverClient(ctx, client, err); ok { client = newClient if isPlainMedia { imgData, err = client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, "m", downloadOptions) @@ -58,6 +58,7 @@ func (h *Handler) ConvertImage(ctx context.Context, portal *bridgev2.Portal, int imgData, err = client.DownloadOBSWithOptions(ctx, oid, talkMetaMessageID, downloadOptions) } } + h.handleFinalAuthError(ctx, client, err) downloadDuration := time.Since(dlStart) if err != nil { diff --git a/pkg/connector/handlers/post_notification.go b/pkg/connector/handlers/post_notification.go index fa80627..19a4226 100644 --- a/pkg/connector/handlers/post_notification.go +++ b/pkg/connector/handlers/post_notification.go @@ -211,15 +211,17 @@ func (h *Handler) convertAlbumPreview( previewContext.ChatID, previewContext.AlbumID, ) - if newClient, ok := h.tryRecoverClient(ctx, err); ok { + if newClient, ok := h.tryRecoverClient(ctx, client, err); ok { + client = newClient imageData, err = h.downloadAlbumPreview( ctx, - newClient, + client, media.OID, previewContext.ChatID, previewContext.AlbumID, ) } + h.handleFinalAuthError(ctx, client, err) if errors.Is(err, line.ErrOBSObjectNotFound) { h.Log.Warn(). Str("msg_id", messageID). diff --git a/pkg/connector/handlers/post_notification_test.go b/pkg/connector/handlers/post_notification_test.go index ffe33fb..035e16d 100644 --- a/pkg/connector/handlers/post_notification_test.go +++ b/pkg/connector/handlers/post_notification_test.go @@ -310,12 +310,6 @@ func TestConvertPostNotificationCancelsQueuedAlbumPreviewsAfterFailure(t *testin NewClient: func() *line.Client { return line.NewClient("token") }, - IsLoggedOut: func(error) bool { - return false - }, - ShouldRecover: func(context.Context, error) bool { - return false - }, DownloadAlbumPreview: func(ctx context.Context, _ *line.Client, oid, _, _ string) ([]byte, error) { calls.Add(1) if oid == "fatal-oid" { @@ -397,12 +391,6 @@ func TestConvertPostNotificationKeepsSuccessfulAlbumImagesWhenOneExpired(t *test NewClient: func() *line.Client { return line.NewClient("token") }, - IsLoggedOut: func(error) bool { - return false - }, - ShouldRecover: func(context.Context, error) bool { - return false - }, DownloadAlbumPreview: func(_ context.Context, _ *line.Client, oid, _, _ string) ([]byte, error) { if oid == "expired-oid" { return nil, line.ErrOBSObjectNotFound @@ -440,12 +428,6 @@ func TestConvertPostNotificationLeavesTransientAlbumFailureRetryable(t *testing. NewClient: func() *line.Client { return line.NewClient("token") }, - IsLoggedOut: func(error) bool { - return false - }, - ShouldRecover: func(context.Context, error) bool { - return false - }, DownloadAlbumPreview: func(context.Context, *line.Client, string, string, string) ([]byte, error) { return nil, line.ErrOBSEncodingIncomplete }, diff --git a/pkg/connector/handlers/video.go b/pkg/connector/handlers/video.go index f9195a1..1aca568 100644 --- a/pkg/connector/handlers/video.go +++ b/pkg/connector/handlers/video.go @@ -50,10 +50,11 @@ func (h *Handler) ConvertVideo(ctx context.Context, portal *bridgev2.Portal, int dlStart := time.Now() videoData, err := client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, sid, downloadOptions) - if newClient, ok := h.tryRecoverClient(ctx, err); ok { + if newClient, ok := h.tryRecoverClient(ctx, client, err); ok { client = newClient videoData, err = client.DownloadOBSWithSIDOptions(ctx, oid, talkMetaMessageID, sid, downloadOptions) } + h.handleFinalAuthError(ctx, client, err) if err != nil { h.Log.Warn(). diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 1239219..af3a984 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -117,14 +117,9 @@ var ( ) func (lc *LineClient) getMessageBoxesWithRecovery(ctx context.Context, opts line.MessageBoxesOptions) (*line.MessageBoxesResponse, error) { - client := lc.newClient() - res, err := client.GetMessageBoxes(opts) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - res, err = client.GetMessageBoxes(opts) - } - } + _, res, err := callLineResult(lc, ctx, func(client *line.Client) (*line.MessageBoxesResponse, error) { + return client.GetMessageBoxes(opts) + }) return res, err } @@ -157,14 +152,9 @@ func (lc *LineClient) fetchAllMessageBoxes(ctx context.Context, opts line.Messag } func (lc *LineClient) refreshBlockedContacts(ctx context.Context) ([]string, error) { - client := lc.newClient() - blockedMIDs, err := client.GetBlockedContactIds() - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - blockedMIDs, err = client.GetBlockedContactIds() - } - } + _, blockedMIDs, err := callLineResult(lc, ctx, func(client *line.Client) ([]string, error) { + return client.GetBlockedContactIds() + }) if err != nil { return nil, err } @@ -600,14 +590,9 @@ func (lc *LineClient) FetchMessages(ctx context.Context, params bridgev2.FetchMe limit = 50 } - client := lc.newClient() - msgs, err := client.GetRecentMessagesV2(chatMID, limit) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - msgs, err = client.GetRecentMessagesV2(chatMID, limit) - } - } + _, msgs, err := callLineResult(lc, ctx, func(client *line.Client) ([]*line.Message, error) { + return client.GetRecentMessagesV2(chatMID, limit) + }) if err != nil { if unblockState != nil { unblockState.complete() @@ -771,14 +756,9 @@ func collectStartupBackfillChatMIDs(messageBoxes []line.MessageBox, memberChatMI // (live) message path. Used by prefetchMessages on startup. func (lc *LineClient) backfillRecentMessages(ctx context.Context, chatMID string, limit int) bool { start := time.Now() - client := lc.newClient() - msgs, err := client.GetRecentMessagesV2(chatMID, limit) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - msgs, err = client.GetRecentMessagesV2(chatMID, limit) - } - } + _, msgs, err := callLineResult(lc, ctx, func(client *line.Client) ([]*line.Message, error) { + return client.GetRecentMessagesV2(chatMID, limit) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chatMID).Msg("Failed to fetch recent messages") return false @@ -845,14 +825,9 @@ func (lc *LineClient) syncChats(ctx context.Context) { } func (lc *LineClient) syncChatsNow(ctx context.Context) { - client := lc.newClient() - midsResp, err := client.GetAllChatMids(true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - midsResp, err = client.GetAllChatMids(true, true) - } - } + client, midsResp, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetAllChatMidsResponse, error) { + return client.GetAllChatMids(true, true) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch all chat mids") return @@ -877,13 +852,10 @@ func (lc *LineClient) syncChatsNow(ctx context.Context) { end = len(allMids) } batch := allMids[i:end] - chatsResp, err := client.GetChats(batch, true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - chatsResp, err = client.GetChats(batch, true, true) - } - } + var chatsResp *line.GetChatsResponse + client, chatsResp, err = callLineResultUsing(lc, ctx, client, func(client *line.Client) (*line.GetChatsResponse, error) { + return client.GetChats(batch, true, true) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to fetch batch of chats") continue @@ -1210,14 +1182,9 @@ func (lc *LineClient) cacheGroupMembersFromRecentMessages(ctx context.Context, c if len(lc.getCachedGroupMembers(chatMid)) > 1 { return } - client := lc.newClient() - msgs, err := client.GetRecentMessagesV2(chatMid, 50) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - msgs, err = client.GetRecentMessagesV2(chatMid, 50) - } - } + _, msgs, err := callLineResult(lc, ctx, func(client *line.Client) ([]*line.Message, error) { + return client.GetRecentMessagesV2(chatMid, 50) + }) if err != nil { lc.UserLogin.Bridge.Log.Debug().Err(err).Str("chat_mid", chatMid).Msg("Failed to fetch recent messages for group member cache") return @@ -1552,20 +1519,13 @@ func (lc *LineClient) pollLoop(ctx context.Context) { client := lc.newClient() lc.UserLogin.Bridge.Log.Info().Msg("Starting LINE SSE loop...") - rev, err := getLastOpRevisionWithClient(ctx, client) - if err != nil && lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) - return - } - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - rev, err = getLastOpRevisionWithClient(ctx, client) - } else { - lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token for getLastOpRevision") - } - } + _, rev, err := callLineResultUsing(lc, ctx, client, func(client *line.Client) (int64, error) { + return getLastOpRevisionWithClient(ctx, client) + }) if err != nil { + if lc.isSessionInvalidated() { + return + } lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to get last op revision") } else { localRev = rev @@ -1677,13 +1637,8 @@ func (lc *LineClient) pollLoop(ctx context.Context) { continue } - if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) - return - } - - if line.IsUnauthorizedStatus(err) { - if lc.handleReceiveAuthError(ctx, err) { + if line.IsAuthError(err) { + if lc.handleReceiveAuthError(ctx, client, err) { return } } @@ -1708,7 +1663,8 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { // This is only a health probe. Keep localRev unchanged so the reconnected // stream replays operations that arrived while the old stream was stalled. - _, probeErr := getLastOpRevisionWithClient(ctx, lc.newClient()) + probeClient := lc.newClient() + _, probeErr := getLastOpRevisionWithClient(ctx, probeClient) if probeErr == nil { return false } @@ -1716,27 +1672,26 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { return true } - if lc.isLoggedOut(probeErr) { - lc.markLoggedOutByOtherClient(ctx, probeErr) - return true - } - if line.IsUnauthorizedStatus(probeErr) { - return lc.handleReceiveAuthError(ctx, probeErr) + return lc.handleReceiveAuthError(ctx, probeClient, probeErr) } - if lc.shouldAttemptTokenRecovery(ctx, probeErr) { - if errRecover := lc.recoverToken(ctx); errRecover != nil { - if errors.Is(errRecover, errLineSessionInvalidated) || lc.isLoggedOut(errRecover) { - lc.markLoggedOutByOtherClient(ctx, errRecover) - return true - } - if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { - lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token after receive auth probe") - } + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, probeClient, probeErr) + if errRecover != nil { + if errors.Is(errRecover, errLineSessionInvalidated) || errors.Is(errRecover, errLineClientSuperseded) || ctx.Err() != nil { + return true + } + if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { + lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token after receive auth probe") } return false } + if recoveredClient != nil { + return false + } + if lc.isSessionInvalidated() { + return true + } if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(probeErr).Msg("Receive auth probe failed; reconnecting SSE") @@ -1747,50 +1702,60 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { // handleReceiveAuthError handles auth failures from /operation/receive. The // receive endpoint may return only a bare 401/403, so probe getProfile to reveal // the detailed forced-logout envelope before deciding whether recovery is safe. -func (lc *LineClient) handleReceiveAuthError(ctx context.Context, err error) bool { +func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient *line.Client, err error) bool { if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) - return true + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + if errRecover != nil { + return true + } + return recoveredClient == nil } - _, profileErr := getProfileWithToken(ctx, lc.getAccessToken()) + profileToken := lc.getAccessToken() + if failedClient != nil && failedClient.AccessToken != "" { + profileToken = failedClient.AccessToken + } + _, profileErr := getProfileWithToken(ctx, profileToken) if ctx.Err() != nil { return true } if lc.isLoggedOut(profileErr) { - lc.markLoggedOutByOtherClient(ctx, profileErr) - return true + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, profileErr) + if errRecover != nil { + return true + } + return recoveredClient == nil } if profileErr == nil { return false } - if !lc.shouldAttemptTokenRecovery(ctx, err) { + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + if recoveredClient != nil { + return false + } + if errRecover == nil { return true } - if errRecover := lc.recoverToken(ctx); errRecover != nil { - if ctx.Err() != nil { - return true - } - if errors.Is(errRecover, errLineSessionInvalidated) || lc.isLoggedOut(errRecover) { - lc.markLoggedOutByOtherClient(ctx, errRecover) - return true - } - if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { - lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") - } - if lc.UserLogin != nil && lc.UserLogin.BridgeState != nil { - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateBadCredentials, - Error: "line-logged-out", - Message: "LINE session was invalidated (logged out by another client). Please re-authenticate the bridge.", - UserAction: status.UserActionRelogin, - }) - } + if ctx.Err() != nil { return true } - return false + if errors.Is(errRecover, errLineSessionInvalidated) || errors.Is(errRecover, errLineClientSuperseded) { + return true + } + if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { + lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") + } + if lc.UserLogin != nil && lc.UserLogin.BridgeState != nil { + lc.UserLogin.BridgeState.Send(status.BridgeState{ + StateEvent: status.StateBadCredentials, + Error: "line-logged-out", + Message: "LINE session was invalidated (logged out by another client). Please re-authenticate the bridge.", + UserAction: status.UserActionRelogin, + }) + } + return true } func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { @@ -2141,14 +2106,9 @@ func (lc *LineClient) handleReactionRemove(op line.Operation, chatMid string, se func (lc *LineClient) syncSingleChat(ctx context.Context, op line.Operation) { chatMid := op.Param1 - client := lc.newClient() - chatsResp, err := client.GetChats([]string{chatMid}, true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - chatsResp, err = client.GetChats([]string{chatMid}, true, true) - } - } + _, chatsResp, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetChatsResponse, error) { + return client.GetChats([]string{chatMid}, true, true) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chatMid).Msg("Failed to fetch chat info") // Only emit leave if we confirm the user is definitively not a member @@ -2202,14 +2162,9 @@ func (lc *LineClient) syncSingleChat(ctx context.Context, op line.Operation) { // checkChatMembership calls GetAllChatMids to verify whether the bridge user // is a member or invitee of the given chat. func (lc *LineClient) checkChatMembership(ctx context.Context, chatMid string) (isMember, isInvitee bool) { - client := lc.newClient() - midsResp, err := client.GetAllChatMids(true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - midsResp, err = client.GetAllChatMids(true, true) - } - } + _, midsResp, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetAllChatMidsResponse, error) { + return client.GetAllChatMids(true, true) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("checkChatMembership: GetAllChatMids failed") return false, false @@ -2329,14 +2284,9 @@ func (lc *LineClient) handleMemberJoin(chatMid, joinerMid string) { } func (lc *LineClient) handleInvite(ctx context.Context, chatMid string, opType OperationType) { - client := lc.newClient() - chatsResp, err := client.GetChats([]string{chatMid}, true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - chatsResp, err = client.GetChats([]string{chatMid}, true, true) - } - } + _, chatsResp, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetChatsResponse, error) { + return client.GetChats([]string{chatMid}, true, true) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chatMid).Msg("Failed to fetch chat info for invite") return @@ -2377,14 +2327,9 @@ func (lc *LineClient) handleInvite(ctx context.Context, chatMid string, opType O } func (lc *LineClient) handleInviteForSelf(ctx context.Context, chatMid string) { - client := lc.newClient() - chatsResp, err := client.GetChats([]string{chatMid}, true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - chatsResp, err = client.GetChats([]string{chatMid}, true, true) - } - } + _, chatsResp, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetChatsResponse, error) { + return client.GetChats([]string{chatMid}, true, true) + }) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chatMid).Msg("Failed to fetch invited chat info") return diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index d16b7c8..1d99728 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -848,7 +848,7 @@ func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) { } lc := &LineClient{AccessToken: "stale"} - stopped := lc.handleReceiveAuthError(context.Background(), errors.New(`SSE error: 401: {"code":10004,"message":"REQUEST_NEED_LOGIN"}`)) + stopped := lc.handleReceiveAuthError(context.Background(), line.NewClient("stale"), errors.New(`SSE error: 401: {"code":10004,"message":"REQUEST_NEED_LOGIN"}`)) if !stopped { t.Fatal("receive auth handler should stop on REQUEST_NEED_LOGIN") @@ -877,7 +877,7 @@ func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { } lc := &LineClient{AccessToken: "valid"} - stopped := lc.handleReceiveAuthError(context.Background(), errors.New("SSE error: 401")) + stopped := lc.handleReceiveAuthError(context.Background(), line.NewClient("valid"), errors.New("SSE error: 401")) if stopped { t.Fatal("receive auth handler should reconnect without stopping when the profile probe succeeds") @@ -893,6 +893,35 @@ func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { } } +func TestReceiveAuthErrorFromStaleSSEClientReconnectsCurrentToken(t *testing.T) { + oldGetProfile := getProfileWithToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + }) + + var profileCalls int + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { + profileCalls++ + if token != "old-token" { + t.Fatalf("profile token = %q, want failed SSE token", token) + } + return nil, errLoggedOut + } + + lc := &LineClient{AccessToken: "current-token"} + stopped := lc.handleReceiveAuthError(context.Background(), line.NewClient("old-token"), errors.New("SSE error: 401")) + + if stopped { + t.Fatal("stale SSE auth error stopped the current session") + } + if profileCalls != 1 { + t.Fatalf("profile calls = %d, want 1", profileCalls) + } + if lc.getAccessToken() != "current-token" || lc.isSessionInvalidated() { + t.Fatal("stale SSE auth error changed current session state") + } +} + func TestReceiveAuthErrorCancellationDuringProfileDoesNotInvalidate(t *testing.T) { oldGetProfile := getProfileWithToken t.Cleanup(func() { @@ -913,7 +942,7 @@ func TestReceiveAuthErrorCancellationDuringProfileDoesNotInvalidate(t *testing.T lc := &LineClient{AccessToken: "valid"} result := make(chan bool, 1) go func() { - result <- lc.handleReceiveAuthError(ctx, errors.New("SSE error: 401")) + result <- lc.handleReceiveAuthError(ctx, line.NewClient("valid"), errors.New("SSE error: 401")) }() <-profileStarted cancel() diff --git a/pkg/connector/userinfo.go b/pkg/connector/userinfo.go index b2f6eb6..1d0f2b1 100644 --- a/pkg/connector/userinfo.go +++ b/pkg/connector/userinfo.go @@ -144,14 +144,9 @@ func (lc *LineClient) GetChatInfo(ctx context.Context, portal *bridgev2.Portal) mid := string(portal.ID) lowerMid := strings.ToLower(mid) if strings.HasPrefix(lowerMid, "c") || strings.HasPrefix(lowerMid, "r") { - client := lc.newClient() - res, err := client.GetChats([]string{mid}, true, true) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - res, err = client.GetChats([]string{mid}, true, true) - } - } + _, res, err := callLineResult(lc, ctx, func(client *line.Client) (*line.GetChatsResponse, error) { + return client.GetChats([]string{mid}, true, true) + }) if err != nil { return nil, err } @@ -211,14 +206,9 @@ func (lc *LineClient) getContact(ctx context.Context, mid string) line.Contact { // Use GetProfile for our own user data if mid == lc.Mid || mid == string(lc.UserLogin.ID) { - client := lc.newClient() - profile, err := client.GetProfile() - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - profile, err = client.GetProfile() - } - } + _, profile, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Profile, error) { + return client.GetProfile() + }) if err == nil && profile != nil { contact := line.Contact{Mid: mid, DisplayName: profile.DisplayName, PicturePath: profile.PicturePath} lc.setCachedContact(mid, contact) @@ -228,13 +218,9 @@ func (lc *LineClient) getContact(ctx context.Context, mid string) line.Contact { } client := lc.newClient() - res, err := client.GetContactsV2([]string{mid}) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - res, err = client.GetContactsV2([]string{mid}) - } - } + client, res, err := callLineResultUsing(lc, ctx, client, func(client *line.Client) (*line.ContactsResponse, error) { + return client.GetContactsV2([]string{mid}) + }) if err == nil && res != nil && res.Contacts != nil { if wrapper, ok := res.Contacts[mid]; ok { lc.setCachedContact(mid, wrapper.Contact) @@ -244,13 +230,9 @@ func (lc *LineClient) getContact(ctx context.Context, mid string) line.Contact { // Fall back to BuddyService for official/business accounts lc.UserLogin.Bridge.Log.Debug().Str("mid", mid).Msg("Contact not found via GetContactsV2, trying BuddyService") - buddy, err := client.GetBuddyProfile(mid) - if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - buddy, err = client.GetBuddyProfile(mid) - } - } + _, buddy, err := callLineResultUsing(lc, ctx, client, func(client *line.Client) (*line.BuddyProfile, error) { + return client.GetBuddyProfile(mid) + }) if err == nil && buddy != nil { lc.UserLogin.Bridge.Log.Debug().Str("mid", mid).Str("display_name", buddy.DisplayName).Str("picture_path", buddy.PicturePath).Msg("Got buddy profile") contact := line.Contact{Mid: mid, DisplayName: buddy.DisplayName, PicturePath: buddy.PicturePath} @@ -332,19 +314,12 @@ func (lc *LineClient) SearchUsers(ctx context.Context, query string) ([]*bridgev } // Search contacts by display name - client := lc.newClient() - allMids, err := client.GetAllContactIds() + client, allMids, err := callLineResult(lc, ctx, func(client *line.Client) ([]string, error) { + return client.GetAllContactIds() + }) if err != nil { - if lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - allMids, err = client.GetAllContactIds() - } - } - if err != nil { - lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to get all contact IDs for search") - return results, nil - } + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to get all contact IDs for search") + return results, nil } // Fetch contacts in batches to check display names @@ -378,18 +353,11 @@ func (lc *LineClient) SearchUsers(ctx context.Context, query string) ([]*bridgev var _ bridgev2.UserSearchingNetworkAPI = (*LineClient)(nil) func (lc *LineClient) GetContactList(ctx context.Context) ([]*bridgev2.ResolveIdentifierResponse, error) { - client := lc.newClient() - allMids, err := client.GetAllContactIds() + client, allMids, err := callLineResult(lc, ctx, func(client *line.Client) ([]string, error) { + return client.GetAllContactIds() + }) if err != nil { - if lc.shouldAttemptTokenRecovery(ctx, err) { - if errRecover := lc.recoverToken(ctx); errRecover == nil { - client = lc.newClient() - allMids, err = client.GetAllContactIds() - } - } - if err != nil { - return nil, err - } + return nil, err } var results []*bridgev2.ResolveIdentifierResponse From a5d41d0a1ae68bd026d5810763dc837f66187a7c Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 30 Jul 2026 11:09:04 -0400 Subject: [PATCH 4/4] fix: address auth recovery review feedback --- pkg/connector/client.go | 31 ------------- pkg/connector/forced_logout_test.go | 70 ++++++++++++++++++++++------- pkg/connector/sync.go | 28 ++++++------ pkg/connector/sync_test.go | 30 +++++++++++-- 4 files changed, 94 insertions(+), 65 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 72eda47..b0e3786 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -307,10 +307,6 @@ func (lc *LineClient) refreshAndSave(ctx context.Context) error { return nil } -func (lc *LineClient) isRefreshRequired(err error) bool { - return line.IsRefreshRequired(err) -} - func (lc *LineClient) isLoggedOut(err error) bool { return line.IsLoggedOut(err) } @@ -788,33 +784,6 @@ func (lc *LineClient) ensureValidToken(ctx context.Context) error { return nil } -func (lc *LineClient) ensureValidTokenWith( - ctx context.Context, - profile func(context.Context) error, - refresh func(context.Context) error, - relogin func(context.Context) error, -) error { - err := profile(ctx) - if err == nil { - return nil - } - if ctx.Err() != nil { - return ctx.Err() - } - - if lc.isLoggedOut(err) { - return err - } - - if !lc.isRefreshRequired(err) { - lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("GetProfile failed with non-auth error, continuing anyway") - return nil - } - - lc.UserLogin.Bridge.Log.Info().Msg("Access token expired, attempting refresh...") - return lc.recoverTokenWith(ctx, refresh, relogin) -} - func (lc *LineClient) Disconnect() { // Disconnect is terminal for this NetworkAPI instance. Framework reconnects // create a replacement client, so late handlers on this one must not mutate diff --git a/pkg/connector/forced_logout_test.go b/pkg/connector/forced_logout_test.go index 572d09e..0253999 100644 --- a/pkg/connector/forced_logout_test.go +++ b/pkg/connector/forced_logout_test.go @@ -50,30 +50,57 @@ func TestEnsureValidTokenReturnsLoggedOutWithoutRelogin(t *testing.T) { } func TestEnsureValidTokenDoesNotReloginAfterLoggedOutRefresh(t *testing.T) { + oldGetProfile := getProfileWithToken + oldRecover := recoverLineToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + recoverLineToken = oldRecover + }) + lc := &LineClient{ + AccessToken: "expired", UserLogin: &bridgev2.UserLogin{ Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, }, } var reloginCalls int - err := lc.ensureValidTokenWith( - context.Background(), - func(context.Context) error { return errAuthRequired }, - func(context.Context) error { return errLoggedOut }, - func(context.Context) error { - reloginCalls++ - return nil - }, - ) - if !line.IsLoggedOut(err) { - t.Fatalf("ensureValidTokenWith error = %v, want logged-out error", err) + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { + if token != "expired" { + t.Fatalf("profile token = %q, want expired", token) + } + return nil, errAuthRequired + } + recoverLineToken = func(lc *LineClient, ctx context.Context) error { + return lc.recoverTokenWith( + ctx, + func(context.Context) error { return errLoggedOut }, + func(context.Context) error { + reloginCalls++ + return nil + }, + ) + } + + err := lc.ensureValidToken(context.Background()) + if !line.IsAuthError(err) { + t.Fatalf("ensureValidToken error = %v, want auth error", err) } if reloginCalls != 0 { t.Fatalf("relogin calls = %d, want 0", reloginCalls) } + if lc.hasAccessToken() || !lc.isSessionInvalidated() { + t.Fatal("logged-out refresh did not invalidate the session") + } } func TestForcedLogoutWinsOverEnsureValidTokenRefresh(t *testing.T) { + oldGetProfile := getProfileWithToken + oldRecover := recoverLineToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + recoverLineToken = oldRecover + }) + lc := &LineClient{ AccessToken: "old-token", UserLogin: &bridgev2.UserLogin{ @@ -84,10 +111,18 @@ func TestForcedLogoutWinsOverEnsureValidTokenRefresh(t *testing.T) { allowRefresh := make(chan struct{}) ensureDone := make(chan error, 1) var reloginCalls int - go func() { - ensureDone <- lc.ensureValidTokenWith( - context.Background(), - func(context.Context) error { return errAuthRequired }, + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { + if token == "recovered-token" { + return &line.Profile{}, nil + } + if token != "old-token" { + t.Fatalf("profile token = %q, want old-token or recovered-token", token) + } + return nil, errAuthRequired + } + recoverLineToken = func(lc *LineClient, ctx context.Context) error { + return lc.recoverTokenWith( + ctx, func(context.Context) error { close(refreshStarted) <-allowRefresh @@ -99,6 +134,9 @@ func TestForcedLogoutWinsOverEnsureValidTokenRefresh(t *testing.T) { return nil }, ) + } + go func() { + ensureDone <- lc.ensureValidToken(context.Background()) }() <-refreshStarted @@ -110,7 +148,7 @@ func TestForcedLogoutWinsOverEnsureValidTokenRefresh(t *testing.T) { close(allowRefresh) if err := <-ensureDone; err != nil { - t.Fatalf("ensureValidTokenWith returned error: %v", err) + t.Fatalf("ensureValidToken returned error: %v", err) } if reloginCalls != 0 { t.Fatalf("relogin calls = %d, want 0", reloginCalls) diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index af3a984..376eb69 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -1712,15 +1712,16 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * } profileToken := lc.getAccessToken() - if failedClient != nil && failedClient.AccessToken != "" { + if profileToken == "" && failedClient != nil { profileToken = failedClient.AccessToken } + profileClient := newLineAPIClient(profileToken) _, profileErr := getProfileWithToken(ctx, profileToken) if ctx.Err() != nil { return true } if lc.isLoggedOut(profileErr) { - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, profileErr) + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, profileClient, profileErr) if errRecover != nil { return true } @@ -1944,7 +1945,7 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { // Curr == nil signals a reaction removal/clear from LINE. if param2.Curr == nil { lc.UserLogin.Bridge.Log.Debug().Str("msg_id", op.Param1).Str("chat_mid", param2.ChatMid).Msg("Received reaction removal (self)") - lc.handleReactionRemove(op, param2.ChatMid, []networkid.UserID{makeUserID(string(lc.UserLogin.ID))}) + lc.handleReactionRemove(op, param2.ChatMid, makeUserID(string(lc.UserLogin.ID))) return } @@ -1983,7 +1984,7 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { // use the type 140 actor from param3, so the sender is unambiguous. if param2.Curr == nil { lc.UserLogin.Bridge.Log.Debug().Str("msg_id", op.Param1).Str("chat_mid", param2.ChatMid).Msg("Received reaction removal (other)") - lc.handleReactionRemove(op, param2.ChatMid, []networkid.UserID{makeUserID(op.Param3)}) + lc.handleReactionRemove(op, param2.ChatMid, makeUserID(op.Param3)) return } @@ -2091,17 +2092,14 @@ func (lc *LineClient) liveReactionSyncEvent( } } -// handleReactionRemove queues an authoritative empty reaction sync for each -// candidate sender. LINE only allows one reaction per sender, so this removes -// both legacy empty-ID rows and stable paid/predefined reaction IDs without -// needing the previous reaction type. -func (lc *LineClient) handleReactionRemove(op line.Operation, chatMid string, senders []networkid.UserID) { - for _, sender := range senders { - lc.UserLogin.Bridge.QueueRemoteEvent( - lc.UserLogin, - lc.liveReactionSyncEvent(op, chatMid, sender, nil), - ) - } +// handleReactionRemove queues an authoritative empty reaction sync for the +// sender, removing both legacy empty-ID rows and stable paid/predefined reaction +// IDs without needing the previous reaction type. +func (lc *LineClient) handleReactionRemove(op line.Operation, chatMid string, sender networkid.UserID) { + lc.UserLogin.Bridge.QueueRemoteEvent( + lc.UserLogin, + lc.liveReactionSyncEvent(op, chatMid, sender, nil), + ) } func (lc *LineClient) syncSingleChat(ctx context.Context, op line.Operation) { diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 1d99728..c9bb435 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -902,10 +902,10 @@ func TestReceiveAuthErrorFromStaleSSEClientReconnectsCurrentToken(t *testing.T) var profileCalls int getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { profileCalls++ - if token != "old-token" { - t.Fatalf("profile token = %q, want failed SSE token", token) + if token != "current-token" { + t.Fatalf("profile token = %q, want current token", token) } - return nil, errLoggedOut + return &line.Profile{}, nil } lc := &LineClient{AccessToken: "current-token"} @@ -922,6 +922,30 @@ func TestReceiveAuthErrorFromStaleSSEClientReconnectsCurrentToken(t *testing.T) } } +func TestReceiveAuthErrorFromStaleSSEClientClassifiesCurrentProbeLogout(t *testing.T) { + oldGetProfile := getProfileWithToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + }) + + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { + if token != "current-token" { + t.Fatalf("profile token = %q, want current token", token) + } + return nil, errLoggedOut + } + + lc := &LineClient{AccessToken: "current-token"} + stopped := lc.handleReceiveAuthError(context.Background(), line.NewClient("old-token"), errors.New("SSE error: 401")) + + if !stopped { + t.Fatal("current-token profile logout should stop the session") + } + if lc.hasAccessToken() || !lc.isSessionInvalidated() { + t.Fatal("current-token profile logout was misclassified as a stale SSE response") + } +} + func TestReceiveAuthErrorCancellationDuringProfileDoesNotInvalidate(t *testing.T) { oldGetProfile := getProfileWithToken t.Cleanup(func() {