From 0b8a5852aedba1865d0920efc0602ad62964d3a4 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 9 Jul 2026 12:34:06 -0500 Subject: [PATCH 1/2] fix: /receive now handles login from another location --- pkg/connector/client.go | 17 ++++++--- pkg/connector/sync.go | 52 ++++++++++++++++++------- pkg/connector/sync_test.go | 77 ++++++++++++++++++++++++++++++++++++++ pkg/line/errors.go | 14 ++++++- pkg/line/errors_test.go | 3 ++ pkg/line/sse.go | 7 ++++ pkg/line/sse_test.go | 39 +++++++++++++++++++ 7 files changed, 189 insertions(+), 20 deletions(-) create mode 100644 pkg/line/sse_test.go diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 30c6d39..4be4ecf 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -287,15 +287,20 @@ func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) if lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("LINE session invalidated by another client; marking login bad credentials") } - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateBadCredentials, - Error: "line-logged-out", - Message: "LINE logged this Chrome Extension session out because another LINE client connected. Click Reconnect in Beeper to reconnect LINE.", - UserAction: status.UserActionRelogin, - }) + if lc.UserLogin.BridgeState != nil { + lc.UserLogin.BridgeState.Send(status.BridgeState{ + StateEvent: status.StateBadCredentials, + Error: "line-logged-out", + Message: "LINE logged this Chrome Extension session out because another LINE client connected. Click Reconnect in Beeper to reconnect LINE.", + UserAction: status.UserActionRelogin, + }) + } } func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { + if lc.UserLogin == nil || lc.UserLogin.UserLogin == nil { + return + } meta, ok := lc.UserLogin.Metadata.(*UserLoginMetadata) if !ok { return diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 93fc6f7..6384ec3 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -1247,19 +1247,8 @@ func (lc *LineClient) pollLoop(ctx context.Context) { return } - if strings.Contains(err.Error(), "SSE error: 401") || - strings.Contains(err.Error(), "SSE error: 403") { - if !lc.shouldAttemptTokenRecovery(ctx, err) { - return - } - if errRecover := lc.recoverToken(ctx); errRecover != nil { - lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") - 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 line.IsUnauthorizedStatus(err) { + if lc.handleReceiveAuthError(ctx, err) { return } } @@ -1269,6 +1258,43 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } } +// 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 { + if lc.isLoggedOut(err) { + lc.markLoggedOutByOtherClient(ctx, err) + return true + } + + if _, profileErr := getProfileWithToken(lc.getAccessToken()); lc.isLoggedOut(profileErr) { + lc.markLoggedOutByOtherClient(ctx, profileErr) + return true + } + + if !lc.shouldAttemptTokenRecovery(ctx, err) { + return true + } + + if errRecover := lc.recoverToken(ctx); errRecover != nil { + if errors.Is(errRecover, errLineSessionInvalidated) || lc.isLoggedOut(errRecover) { + lc.markLoggedOutByOtherClient(ctx, errRecover) + return true + } + lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") + if 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 + } + return false +} + func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { opType := OperationType(op.Type) diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index b220e9d..61504f9 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -2,6 +2,7 @@ package connector import ( "context" + "errors" "io" "testing" "time" @@ -61,3 +62,79 @@ func TestPollLoopRebuildsSSEClientAfterReconnect(t *testing.T) { t.Fatalf("SSE tokens = %v, want [old new]", tokens) } } + +func TestPollLoopMarksLoggedOutWhenReceiveAuthFails(t *testing.T) { + oldGetLastOpRevision := getLastOpRevisionWithClient + oldListenSSE := listenSSEWithClient + oldGetProfile := getProfileWithToken + t.Cleanup(func() { + getLastOpRevisionWithClient = oldGetLastOpRevision + listenSSEWithClient = oldListenSSE + getProfileWithToken = oldGetProfile + }) + + getLastOpRevisionWithClient = func(*line.Client) (int64, error) { + return 1234, nil + } + + var profileCalls int + getProfileWithToken = func(token string) (*line.Profile, error) { + profileCalls++ + if token != "stale" { + t.Fatalf("profile token = %q, want stale", token) + } + return nil, errLoggedOut + } + + listenSSEWithClient = func(client *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error { + if client.AccessToken != "stale" { + t.Fatalf("SSE client token = %q, want stale", client.AccessToken) + } + return errors.New("SSE error: 401") + } + + lc := &LineClient{ + AccessToken: "stale", + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + + lc.wg.Add(1) + lc.pollLoop(context.Background()) + + if profileCalls != 1 { + t.Fatalf("profile calls = %d, want 1", profileCalls) + } + if lc.hasAccessToken() { + t.Fatal("access token was not invalidated after receive auth logout") + } + if !lc.isSessionInvalidated() { + t.Fatal("session was not marked invalidated after receive auth logout") + } +} + +func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) { + oldGetProfile := getProfileWithToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + }) + + getProfileWithToken = func(token string) (*line.Profile, error) { + t.Fatal("REQUEST_NEED_LOGIN should be handled without probing profile") + return nil, nil + } + + lc := &LineClient{AccessToken: "stale"} + stopped := lc.handleReceiveAuthError(context.Background(), errors.New(`SSE error: 401: {"code":10004,"message":"REQUEST_NEED_LOGIN"}`)) + + if !stopped { + t.Fatal("receive auth handler should stop on REQUEST_NEED_LOGIN") + } + if lc.hasAccessToken() { + t.Fatal("access token was not invalidated") + } + if !lc.isSessionInvalidated() { + t.Fatal("session was not marked invalidated") + } +} diff --git a/pkg/line/errors.go b/pkg/line/errors.go index c99ccf9..8a8bc32 100644 --- a/pkg/line/errors.go +++ b/pkg/line/errors.go @@ -28,7 +28,9 @@ func IsLoggedOut(err error) bool { if err == nil { return false } - return strings.Contains(err.Error(), "V3_TOKEN_CLIENT_LOGGED_OUT") || IsInvalidSenderKey(err) + return strings.Contains(err.Error(), "V3_TOKEN_CLIENT_LOGGED_OUT") || + IsInvalidSenderKey(err) || + IsRequestNeedLogin(err) } func IsInvalidSenderKey(err error) bool { @@ -42,6 +44,16 @@ func IsInvalidSenderKey(err error) bool { strings.Contains(msg, "invalid sender key") } +func IsRequestNeedLogin(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "request_need_login") || + strings.Contains(msg, "\"code\":10004") || + strings.Contains(msg, "\"code\": 10004") +} + func IsUnauthorizedStatus(err error) bool { if err == nil { return false diff --git a/pkg/line/errors_test.go b/pkg/line/errors_test.go index 372e850..4f1200b 100644 --- a/pkg/line/errors_test.go +++ b/pkg/line/errors_test.go @@ -54,6 +54,9 @@ func TestIsLoggedOut(t *testing.T) { if !IsLoggedOut(errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","message":"TalkException","code":83,"reason":"invalid sender key","parameterMap":null}}`)) { t.Fatal("expected invalid sender key to be detected as logged out") } + if !IsLoggedOut(errors.New(`SSE error: 401: {"code":10004,"message":"REQUEST_NEED_LOGIN"}`)) { + t.Fatal("expected request-need-login to be detected as logged out") + } if IsLoggedOut(errors.New("Access token refresh required")) { t.Fatal("refresh-required error should not be classified as logged out") } diff --git a/pkg/line/sse.go b/pkg/line/sse.go index d350b49..d7373e9 100644 --- a/pkg/line/sse.go +++ b/pkg/line/sse.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "fmt" + "io" "net/http" "net/url" "strconv" @@ -14,6 +15,8 @@ import ( // Reused across reconnects to avoid allocating a new transport pool each time. var sseHTTPClient = &http.Client{} +const maxSSEErrorBodyBytes = 4096 + // ListenSSE connects to the Event Stream and blocks func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(event, data string)) error { q := url.Values{} @@ -45,6 +48,10 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev defer resp.Body.Close() if resp.StatusCode != 200 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, maxSSEErrorBodyBytes)) + if bodyText := strings.TrimSpace(string(body)); bodyText != "" { + return fmt.Errorf("SSE error: %d: %s", resp.StatusCode, bodyText) + } return fmt.Errorf("SSE error: %d", resp.StatusCode) } diff --git a/pkg/line/sse_test.go b/pkg/line/sse_test.go new file mode 100644 index 0000000..20d556d --- /dev/null +++ b/pkg/line/sse_test.go @@ -0,0 +1,39 @@ +package line + +import ( + "context" + "io" + "net/http" + "strings" + "testing" +) + +func TestListenSSENonOKIncludesResponseBody(t *testing.T) { + oldClient := sseHTTPClient + t.Cleanup(func() { + sseHTTPClient = oldClient + }) + + sseHTTPClient = &http.Client{ + Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader( + `{"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":8,"reason":"V3_TOKEN_CLIENT_LOGGED_OUT"}}`, + )), + }, nil + }), + } + + err := NewClient("stale-token").ListenSSE(context.Background(), 0, func(event, data string) {}) + if err == nil { + t.Fatal("expected SSE error") + } + if !strings.Contains(err.Error(), "SSE error: 401") { + t.Fatalf("err = %v, want status code", err) + } + if !strings.Contains(err.Error(), "V3_TOKEN_CLIENT_LOGGED_OUT") { + t.Fatalf("err = %v, want response body detail", err) + } +} From 49433606d82cc42b68cb47806e7c5c2ed2669125 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 9 Jul 2026 13:01:07 -0500 Subject: [PATCH 2/2] fix: linked 10004 to REQUEST_NEED_LOGIN + reconnect if getProfile succeeds --- pkg/connector/sync.go | 12 +++++++++--- pkg/connector/sync_test.go | 32 ++++++++++++++++++++++++++++++++ pkg/line/errors.go | 17 +++++++++++++++-- pkg/line/errors_test.go | 6 ++++++ 4 files changed, 62 insertions(+), 5 deletions(-) diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 6384ec3..e03eb3b 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -1267,10 +1267,14 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, err error) boo return true } - if _, profileErr := getProfileWithToken(lc.getAccessToken()); lc.isLoggedOut(profileErr) { + _, profileErr := getProfileWithToken(lc.getAccessToken()) + if lc.isLoggedOut(profileErr) { lc.markLoggedOutByOtherClient(ctx, profileErr) return true } + if profileErr == nil { + return false + } if !lc.shouldAttemptTokenRecovery(ctx, err) { return true @@ -1281,8 +1285,10 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, err error) boo lc.markLoggedOutByOtherClient(ctx, errRecover) return true } - lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") - if lc.UserLogin.BridgeState != nil { + 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", diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 61504f9..cccd9cf 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -138,3 +138,35 @@ func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) { t.Fatal("session was not marked invalidated") } } + +func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { + oldGetProfile := getProfileWithToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + }) + + var profileCalls int + getProfileWithToken = func(token string) (*line.Profile, error) { + profileCalls++ + if token != "valid" { + t.Fatalf("profile token = %q, want valid", token) + } + return &line.Profile{}, nil + } + + lc := &LineClient{AccessToken: "valid"} + stopped := lc.handleReceiveAuthError(context.Background(), errors.New("SSE error: 401")) + + if stopped { + t.Fatal("receive auth handler should reconnect without stopping when the profile probe succeeds") + } + if profileCalls != 1 { + t.Fatalf("profile calls = %d, want 1", profileCalls) + } + if !lc.hasAccessToken() { + t.Fatal("valid access token was invalidated") + } + if lc.isSessionInvalidated() { + t.Fatal("valid session was marked invalidated") + } +} diff --git a/pkg/line/errors.go b/pkg/line/errors.go index 8a8bc32..db135e7 100644 --- a/pkg/line/errors.go +++ b/pkg/line/errors.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "strconv" "strings" ) @@ -50,8 +51,7 @@ func IsRequestNeedLogin(err error) bool { } msg := strings.ToLower(err.Error()) return strings.Contains(msg, "request_need_login") || - strings.Contains(msg, "\"code\":10004") || - strings.Contains(msg, "\"code\": 10004") + hasJSONCode(msg, 10004) } func IsUnauthorizedStatus(err error) bool { @@ -201,6 +201,19 @@ func hasResponseErrorCode(msg string) bool { strings.Contains(msg, "code 10051") } +func hasJSONCode(msg string, code int) bool { + value := strconv.Itoa(code) + for _, prefix := range []string{`"code":`, `"code": `} { + codeStart := prefix + value + for _, suffix := range []string{",", "}", " ", "\t", "\r", "\n"} { + if strings.Contains(msg, codeStart+suffix) { + return true + } + } + } + return false +} + func isNoUsableE2EEGroupKeyTalkException(message string, data talkExceptionData) bool { if !strings.EqualFold(message, "RESPONSE_ERROR") || !strings.EqualFold(data.Name, "TalkException") { return false diff --git a/pkg/line/errors_test.go b/pkg/line/errors_test.go index 4f1200b..c03a05f 100644 --- a/pkg/line/errors_test.go +++ b/pkg/line/errors_test.go @@ -57,6 +57,12 @@ func TestIsLoggedOut(t *testing.T) { if !IsLoggedOut(errors.New(`SSE error: 401: {"code":10004,"message":"REQUEST_NEED_LOGIN"}`)) { t.Fatal("expected request-need-login to be detected as logged out") } + if !IsLoggedOut(errors.New(`SSE error: 401: {"code":10004}`)) { + t.Fatal("expected request-need-login code to be detected as logged out") + } + if IsLoggedOut(errors.New(`SSE error: 401: {"code":100040}`)) { + t.Fatal("similar longer numeric code should not be classified as logged out") + } if IsLoggedOut(errors.New("Access token refresh required")) { t.Fatal("refresh-required error should not be classified as logged out") }