fix: receive loop could break if re-auth was called more than once at…

pkg/connector/sync.go

@@ -34,6 +34,16 @@ const (
 	beeperExcludeFromTimelineKey = "com.beeper.exclude_from_timeline"
 )
 
+var (
+	getLastOpRevisionWithClient = func(client *line.Client) (int64, error) {
+		return client.GetLastOpRevision()
+	}
+	listenSSEWithClient = func(client *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error {
+		return client.ListenSSE(ctx, localRev, handler)
+	}
+	sseReconnectDelay = 3 * time.Second
+)
+
 func (lc *LineClient) getMessageBoxesWithRecovery(ctx context.Context, opts line.MessageBoxesOptions) (*line.MessageBoxesResponse, error) {
 	client := lc.newClient()
 	res, err := client.GetMessageBoxes(opts)
@@ -1150,15 +1160,15 @@ func (lc *LineClient) pollLoop(ctx context.Context) {
 	client := lc.newClient()
 
 	lc.UserLogin.Bridge.Log.Info().Msg("Starting LINE SSE loop...")
-	rev, err := client.GetLastOpRevision()
+	rev, err := getLastOpRevisionWithClient(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 = client.GetLastOpRevision()
+			rev, err = getLastOpRevisionWithClient(client)
 		} else {
 			lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token for getLastOpRevision")
 		}
@@ -1223,7 +1233,8 @@ func (lc *LineClient) pollLoop(ctx context.Context) {
 	}
 
 	for {
-		err := client.ListenSSE(ctx, localRev, handler)
+		client = lc.newClient()
+		err := listenSSEWithClient(client, ctx, localRev, handler)
 		if err != nil {
 			if errors.Is(err, context.Canceled) {
 				return
@@ -1251,10 +1262,9 @@ func (lc *LineClient) pollLoop(ctx context.Context) {
 						})
 						return
 					}
-					client = lc.newClient()
 				}
 			}
-			time.Sleep(3 * time.Second)
+			time.Sleep(sseReconnectDelay)
 		}
 	}
 }

pkg/connector/sync_test.go

@@ -0,0 +1,63 @@
+package connector
+
+import (
+	"context"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+	"maunium.net/go/mautrix/bridgev2"
+
+	"github.com/highesttt/matrix-line-messenger/pkg/line"
+)
+
+func TestPollLoopRebuildsSSEClientAfterReconnect(t *testing.T) {
+	oldGetLastOpRevision := getLastOpRevisionWithClient
+	oldListenSSE := listenSSEWithClient
+	oldReconnectDelay := sseReconnectDelay
+	t.Cleanup(func() {
+		getLastOpRevisionWithClient = oldGetLastOpRevision
+		listenSSEWithClient = oldListenSSE
+		sseReconnectDelay = oldReconnectDelay
+	})
+
+	getLastOpRevisionWithClient = func(*line.Client) (int64, error) {
+		return 1234, nil
+	}
+	sseReconnectDelay = time.Millisecond
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	lc := &LineClient{
+		AccessToken: "old",
+		UserLogin: &bridgev2.UserLogin{
+			Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)},
+		},
+	}
+
+	var tokens []string
+	listenSSEWithClient = func(client *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error {
+		if localRev != 1234 {
+			t.Fatalf("localRev = %d, want 1234", localRev)
+		}
+		tokens = append(tokens, client.AccessToken)
+		if len(tokens) == 1 {
+			lc.setTokens("new", "")
+			return io.EOF
+		}
+		cancel()
+		return context.Canceled
+	}
+
+	lc.wg.Add(1)
+	lc.pollLoop(ctx)
+
+	if len(tokens) != 2 {
+		t.Fatalf("SSE attempts = %d, want 2", len(tokens))
+	}
+	if tokens[0] != "old" || tokens[1] != "new" {
+		t.Fatalf("SSE tokens = %v, want [old new]", tokens)
+	}
+}

pkg/runner.go

@@ -481,20 +481,13 @@ func (r *Runner) ChannelCreate(keyID int, peerPublicB64 string) (int, error) {
 
 	id := r.putChannel(chanPtr)
 
-	// If we have raw key material, also create a pure Go channel
+	// If we have raw key material, also create a pure Go channel. This cache is
+	// best-effort; the WASM channel above is the authoritative channel.
 	if goKey, ok := r.goKeys[keyID]; ok && goKey.privKey != nil {
 		goChan, err := ltsm.NewChannel(goKey.privKey, peerPubBytes)
 		if err == nil {
 			r.goChannels[id] = goChan
-		} else {
-			fmt.Printf("DEBUG ChannelCreate: NewChannel failed: %v (keyID=%d len(privKey)=%d)\n", err, keyID, len(goKey.privKey))
 		}
-	} else {
-		fmt.Printf("DEBUG ChannelCreate: no goKey for keyID=%d (goKeys has keys: ", keyID)
-		for k := range r.goKeys {
-			fmt.Printf("%d ", k)
-		}
-		fmt.Printf(")\n")
 	}
 
 	return id, nil

pkg/runner_channel_test.go

@@ -0,0 +1,82 @@
+package line
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"testing"
+)
+
+func TestChannelCreateWithWASMOnlyKeyDoesNotWriteStdout(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping runner integration test in short mode")
+	}
+
+	r, err := GetRunner()
+	if err != nil {
+		t.Fatalf("GetRunner failed: %v", err)
+	}
+
+	keyID, err := r.KeyGenerate()
+	if err != nil {
+		t.Fatalf("KeyGenerate failed: %v", err)
+	}
+	peerKeyID, err := r.KeyGenerate()
+	if err != nil {
+		t.Fatalf("peer KeyGenerate failed: %v", err)
+	}
+	peerPub, err := r.KeyGetPublic(peerKeyID)
+	if err != nil {
+		t.Fatalf("KeyGetPublic failed: %v", err)
+	}
+
+	var channelID int
+	stdout := captureStdout(t, func() {
+		channelID, err = r.ChannelCreate(keyID, peerPub)
+	})
+	if err != nil {
+		t.Fatalf("ChannelCreate failed: %v", err)
+	}
+	if channelID == 0 {
+		t.Fatal("ChannelCreate returned empty channel ID")
+	}
+	if stdout != "" {
+		t.Fatalf("ChannelCreate wrote to stdout: %q", stdout)
+	}
+}
+
+func captureStdout(t *testing.T, fn func()) string {
+	t.Helper()
+
+	oldStdout := os.Stdout
+	readPipe, writePipe, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("os.Pipe failed: %v", err)
+	}
+
+	var buf bytes.Buffer
+	done := make(chan error, 1)
+	go func() {
+		_, copyErr := io.Copy(&buf, readPipe)
+		done <- copyErr
+	}()
+
+	os.Stdout = writePipe
+	defer func() {
+		os.Stdout = oldStdout
+	}()
+	fn()
+	os.Stdout = oldStdout
+
+	if err := writePipe.Close(); err != nil {
+		t.Fatalf("closing stdout pipe writer failed: %v", err)
+	}
+	if err := <-done; err != nil {
+		t.Fatalf("reading captured stdout failed: %v", err)
+	}
+	if err := readPipe.Close(); err != nil {
+		t.Fatalf("closing stdout pipe reader failed: %v", err)
+	}
+
+	return buf.String()
+}