diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 69cb03c..93fc6f7 100644 --- a/pkg/connector/sync.go +++ b/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,7 +1160,7 @@ 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 @@ -1158,7 +1168,7 @@ func (lc *LineClient) pollLoop(ctx context.Context) { 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) } } } diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go new file mode 100644 index 0000000..b220e9d --- /dev/null +++ b/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) + } +} diff --git a/pkg/runner.go b/pkg/runner.go index 77da7b0..147c8c7 100644 --- a/pkg/runner.go +++ b/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 diff --git a/pkg/runner_channel_test.go b/pkg/runner_channel_test.go new file mode 100644 index 0000000..8b390e0 --- /dev/null +++ b/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() +}