fix: /receive now handles login from another location

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

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,49 @@ 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
+	}
+
+	_, 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
+	}
+
+	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.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
+	}
+	return false
+}
+
 func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) {
 	opType := OperationType(op.Type)
 

pkg/connector/sync_test.go

@@ -2,6 +2,7 @@ package connector
 
 import (
 	"context"
+	"errors"
 	"io"
 	"testing"
 	"time"
@@ -61,3 +62,111 @@ 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")
+	}
+}
+
+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")
+	}
+}

pkg/line/errors.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 )
 
@@ -28,7 +29,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 +45,15 @@ 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") ||
+		hasJSONCode(msg, 10004)
+}
+
 func IsUnauthorizedStatus(err error) bool {
 	if err == nil {
 		return false
@@ -189,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

pkg/line/errors_test.go

@@ -54,6 +54,15 @@ 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(`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")
 	}

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)
 	}
 

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)
+	}
+}