From 01b044e3056edb6b587628e7ed0ac9ea84af1cee Mon Sep 17 00:00:00 2001 From: ruinivist Date: Tue, 28 Apr 2026 23:22:51 +0200 Subject: [PATCH] feat: add twitch live detection to streamer list --- internal/app/app.go | 90 ++++++++++++++++----- internal/app/app_test.go | 135 +++++++++++++++++++++++++++++--- internal/gql/operations.go | 18 +++++ internal/tui/model.go | 8 +- internal/tui/model_test.go | 21 ++++- internal/twitch/service.go | 53 +++++++++++++ internal/twitch/service_test.go | 61 +++++++++++++++ 7 files changed, 354 insertions(+), 32 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index f4c558d..da54845 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -21,6 +21,8 @@ import ( "parasocial/internal/twitch" ) +const streamerRefreshInterval = 5 * time.Minute + // Run loads the application configuration and starts the terminal UI. func Run(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) @@ -97,6 +99,8 @@ func Run(ctx context.Context) error { type streamerService interface { CurrentUser(context.Context) (*twitch.Viewer, error) ResolveStreamer(context.Context, string) (*twitch.Channel, error) + StreamInfo(context.Context, string) (*twitch.StreamInfo, error) + PlaybackAccessToken(context.Context, string) (*twitch.PlaybackToken, error) } // newTwitchService builds a Twitch service from the authenticated session state. @@ -118,36 +122,82 @@ func newTwitchService(httpClient *http.Client, state *auth.State) (*twitch.Servi // resolveStreamerEntries streams viewer and streamer resolution results into the TUI. func resolveStreamerEntries(ctx context.Context, service streamerService, logins []string, send func(tui.StreamerUpdate)) error { + return resolveStreamerEntriesWithSleep(ctx, service, logins, send, streamerRefreshInterval, sleepContext) +} + +func resolveStreamerEntriesWithSleep(ctx context.Context, service streamerService, logins []string, send func(tui.StreamerUpdate), interval time.Duration, sleep func(context.Context, time.Duration) error) error { viewer, err := service.CurrentUser(ctx) if err != nil { return fmt.Errorf("resolve current user: %w", err) } send(tui.StreamerUpdate{Viewer: viewer}) - for index, login := range logins { - entry := twitch.StreamerEntry{ - ConfigLogin: login, - Status: twitch.StreamerLoading, + for { + for index, login := range logins { + entry, err := resolveStreamerEntry(ctx, service, login) + if errors.Is(err, context.Canceled) { + return err + } + send(tui.StreamerUpdate{ + Index: index, + Entry: entry, + }) } - - channel, err := service.ResolveStreamer(ctx, login) - switch { - case err == nil: - entry.Login = channel.Login - entry.ChannelID = channel.ID - entry.Status = twitch.StreamerReady - case errors.Is(err, context.Canceled): + if err := sleep(ctx, interval); err != nil { return err - default: - entry.Status = twitch.StreamerError - entry.Error = err.Error() } + } +} - send(tui.StreamerUpdate{ - Index: index, - Entry: &entry, - }) +func resolveStreamerEntry(ctx context.Context, service streamerService, login string) (*twitch.StreamerEntry, error) { + entry := &twitch.StreamerEntry{ + ConfigLogin: login, + Status: twitch.StreamerLoading, } - return nil + channel, err := service.ResolveStreamer(ctx, login) + switch { + case err == nil: + entry.Login = channel.Login + entry.ChannelID = channel.ID + case errors.Is(err, context.Canceled): + return nil, err + default: + entry.Status = twitch.StreamerError + entry.Error = err.Error() + return entry, nil + } + + stream, err := service.StreamInfo(ctx, channel.ID) + switch { + case err == nil: + entry.Status = twitch.StreamerReady + entry.Live = stream.Online + case errors.Is(err, context.Canceled): + return nil, err + default: + entry.Status = twitch.StreamerError + entry.Error = err.Error() + return entry, nil + } + + if !entry.Live { + return entry, nil + } + if _, err := service.PlaybackAccessToken(ctx, channel.Login); err != nil && errors.Is(err, context.Canceled) { + return nil, err + } + return entry, nil +} + +func sleepContext(ctx context.Context, duration time.Duration) error { + timer := time.NewTimer(duration) + defer timer.Stop() + + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } } diff --git a/internal/app/app_test.go b/internal/app/app_test.go index d2af7ac..0ec5d5a 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -7,6 +7,7 @@ import ( "context" "errors" "testing" + "time" "parasocial/internal/tui" "parasocial/internal/twitch" @@ -14,10 +15,18 @@ import ( // fakeStreamerService is a test double for the app's Twitch resolution dependency. type fakeStreamerService struct { - viewer *twitch.Viewer - viewerErr error - channels map[string]*twitch.Channel - errs map[string]error + viewer *twitch.Viewer + viewerErr error + channels map[string]*twitch.Channel + resolveErrs map[string]error + streams map[string][]streamResult + streamCalls map[string]int + playbackErr map[string]error +} + +type streamResult struct { + info *twitch.StreamInfo + err error } // CurrentUser returns the configured fake viewer or viewer error. @@ -30,12 +39,40 @@ func (f *fakeStreamerService) CurrentUser(context.Context) (*twitch.Viewer, erro // ResolveStreamer returns the configured fake channel or resolution error. func (f *fakeStreamerService) ResolveStreamer(_ context.Context, login string) (*twitch.Channel, error) { - if err, ok := f.errs[login]; ok { + if err, ok := f.resolveErrs[login]; ok { return nil, err } return f.channels[login], nil } +func (f *fakeStreamerService) StreamInfo(_ context.Context, channelID string) (*twitch.StreamInfo, error) { + if f.streamCalls == nil { + f.streamCalls = map[string]int{} + } + call := f.streamCalls[channelID] + f.streamCalls[channelID] = call + 1 + + results := f.streams[channelID] + if len(results) == 0 { + return &twitch.StreamInfo{Online: false}, nil + } + if call >= len(results) { + call = len(results) - 1 + } + result := results[call] + if result.err != nil { + return nil, result.err + } + return result.info, nil +} + +func (f *fakeStreamerService) PlaybackAccessToken(_ context.Context, login string) (*twitch.PlaybackToken, error) { + if err, ok := f.playbackErr[login]; ok { + return nil, err + } + return &twitch.PlaybackToken{Signature: "sig", Value: "token"}, nil +} + // TestResolveStreamerEntries verifies that app resolution emits viewer, success, and error updates. func TestResolveStreamerEntries(t *testing.T) { t.Parallel() @@ -45,16 +82,21 @@ func TestResolveStreamerEntries(t *testing.T) { channels: map[string]*twitch.Channel{ "alpha": {ID: "1", Login: "alpha_live"}, }, - errs: map[string]error{ + resolveErrs: map[string]error{ "beta": errors.New("lookup failed"), }, + streams: map[string][]streamResult{ + "1": {{info: &twitch.StreamInfo{Online: true}}}, + }, } var updates []tui.StreamerUpdate - err := resolveStreamerEntries(context.Background(), service, []string{"alpha", "beta"}, func(update tui.StreamerUpdate) { + err := resolveStreamerEntriesWithSleep(context.Background(), service, []string{"alpha", "beta"}, func(update tui.StreamerUpdate) { updates = append(updates, update) + }, 0, func(context.Context, time.Duration) error { + return context.Canceled }) - if err != nil { + if !errors.Is(err, context.Canceled) { t.Fatal(err) } if len(updates) != 3 { @@ -63,10 +105,85 @@ func TestResolveStreamerEntries(t *testing.T) { if updates[0].Viewer == nil || updates[0].Viewer.Login != "viewer" { t.Fatalf("viewer update = %#v", updates[0]) } - if updates[1].Entry == nil || updates[1].Entry.Status != twitch.StreamerReady || updates[1].Entry.Login != "alpha_live" { + if updates[1].Entry == nil || updates[1].Entry.Status != twitch.StreamerReady || updates[1].Entry.Login != "alpha_live" || !updates[1].Entry.Live { t.Fatalf("alpha update = %#v", updates[1]) } if updates[2].Entry == nil || updates[2].Entry.Status != twitch.StreamerError { t.Fatalf("beta update = %#v", updates[2]) } } + +func TestResolveStreamerEntriesPlaybackFailureStillMarksLive(t *testing.T) { + t.Parallel() + + service := &fakeStreamerService{ + viewer: &twitch.Viewer{ID: "7", Login: "viewer"}, + channels: map[string]*twitch.Channel{ + "alpha": {ID: "1", Login: "alpha_live"}, + }, + streams: map[string][]streamResult{ + "1": {{info: &twitch.StreamInfo{Online: true}}}, + }, + playbackErr: map[string]error{ + "alpha_live": errors.New("token lookup failed"), + }, + } + + var updates []tui.StreamerUpdate + err := resolveStreamerEntriesWithSleep(context.Background(), service, []string{"alpha"}, func(update tui.StreamerUpdate) { + updates = append(updates, update) + }, 0, func(context.Context, time.Duration) error { + return context.Canceled + }) + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if len(updates) != 2 { + t.Fatalf("len(updates) = %d", len(updates)) + } + entry := updates[1].Entry + if entry == nil || entry.Status != twitch.StreamerReady || !entry.Live { + t.Fatalf("entry = %#v", entry) + } +} + +func TestResolveStreamerEntriesRefreshesLiveState(t *testing.T) { + t.Parallel() + + service := &fakeStreamerService{ + viewer: &twitch.Viewer{ID: "7", Login: "viewer"}, + channels: map[string]*twitch.Channel{ + "alpha": {ID: "1", Login: "alpha_live"}, + }, + streams: map[string][]streamResult{ + "1": { + {info: &twitch.StreamInfo{Online: false}}, + {info: &twitch.StreamInfo{Online: true}}, + }, + }, + } + + var updates []tui.StreamerUpdate + sleepCalls := 0 + err := resolveStreamerEntriesWithSleep(context.Background(), service, []string{"alpha"}, func(update tui.StreamerUpdate) { + updates = append(updates, update) + }, 0, func(context.Context, time.Duration) error { + sleepCalls++ + if sleepCalls == 1 { + return nil + } + return context.Canceled + }) + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if len(updates) != 3 { + t.Fatalf("len(updates) = %d", len(updates)) + } + if updates[1].Entry == nil || updates[1].Entry.Live { + t.Fatalf("first pass = %#v", updates[1]) + } + if updates[2].Entry == nil || !updates[2].Entry.Live { + t.Fatalf("second pass = %#v", updates[2]) + } +} diff --git a/internal/gql/operations.go b/internal/gql/operations.go index facc136..0bef505 100644 --- a/internal/gql/operations.go +++ b/internal/gql/operations.go @@ -67,3 +67,21 @@ func GetIDFromLogin(login string) Request { "login": login, }) } + +// PlaybackAccessToken fetches the HLS playback token Twitch issues for a live channel. +func PlaybackAccessToken(login string) Request { + return operation("PlaybackAccessToken", "3093517e37e4f4cb48906155bcd894150aef92617939236d2508f3375ab732ce", map[string]any{ + "login": login, + "isLive": true, + "isVod": false, + "vodID": "", + "playerType": "site", + }) +} + +// WithIsStreamLiveQuery fetches live stream metadata for one channel by ID. +func WithIsStreamLiveQuery(channelID string) Request { + return operation("WithIsStreamLiveQuery", "04e46329a6786ff3a81c01c50bfa5d725902507a0deb83b0edbf7abe7a3716ea", map[string]any{ + "id": channelID, + }) +} diff --git a/internal/tui/model.go b/internal/tui/model.go index c5098ac..999ee5c 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -14,6 +14,8 @@ import ( "parasocial/internal/twitch" ) +const liveDot = "\x1b[32m●\x1b[0m" + type viewMode int const ( @@ -195,7 +197,11 @@ func (m Model) View() string { for i, streamer := range m.streamers { switch streamer.Status { case twitch.StreamerReady: - fmt.Fprintf(&builder, "%d. %s (%s)\n", i+1, streamer.Login, streamer.ChannelID) + label := streamer.Login + if streamer.Live { + label += " " + liveDot + } + fmt.Fprintf(&builder, "%d. %s (%s)\n", i+1, label, streamer.ChannelID) case twitch.StreamerError: fmt.Fprintf(&builder, "%d. %s [error: %s]\n", i+1, streamer.ConfigLogin, streamer.Error) default: diff --git a/internal/tui/model_test.go b/internal/tui/model_test.go index c2f663e..cbf6b48 100644 --- a/internal/tui/model_test.go +++ b/internal/tui/model_test.go @@ -10,7 +10,7 @@ import ( func TestViewDisplaysStreamers(t *testing.T) { model := New(Options{ Streamers: []twitch.StreamerEntry{ - {ConfigLogin: "alpha", Login: "alpha_live", ChannelID: "1", Status: twitch.StreamerReady}, + {ConfigLogin: "alpha", Login: "alpha_live", ChannelID: "1", Live: true, Status: twitch.StreamerReady}, {ConfigLogin: "beta", Status: twitch.StreamerLoading}, }, }) @@ -18,7 +18,7 @@ func TestViewDisplaysStreamers(t *testing.T) { model.viewer = &twitch.Viewer{ID: "7", Login: "viewer"} got := model.View() - want := "Logged in as viewer (7)\n\nStreamers\n1. alpha_live (1)\n2. beta [loading]\n\n" + want := "Logged in as viewer (7)\n\nStreamers\n1. alpha_live \x1b[32m●\x1b[0m (1)\n2. beta [loading]\n\n" if got != want { t.Fatalf("View() = %q, want %q", got, want) } @@ -112,6 +112,7 @@ func TestStreamerUpdateAppliesEntry(t *testing.T) { ConfigLogin: "alpha", Login: "alpha_live", ChannelID: "1", + Live: true, Status: twitch.StreamerReady, }, }) @@ -121,6 +122,22 @@ func TestStreamerUpdateAppliesEntry(t *testing.T) { next := updated.(Model) got := next.View() + want := "Logged in as viewer (7)\n\nStreamers\n1. alpha_live \x1b[32m●\x1b[0m (1)\n\n" + if got != want { + t.Fatalf("View() = %q, want %q", got, want) + } +} + +func TestViewDisplaysOfflineReadyStreamerWithoutMarker(t *testing.T) { + model := New(Options{ + Streamers: []twitch.StreamerEntry{ + {ConfigLogin: "alpha", Login: "alpha_live", ChannelID: "1", Status: twitch.StreamerReady}, + }, + }) + model.mode = streamerView + model.viewer = &twitch.Viewer{ID: "7", Login: "viewer"} + + got := model.View() want := "Logged in as viewer (7)\n\nStreamers\n1. alpha_live (1)\n\n" if got != want { t.Fatalf("View() = %q, want %q", got, want) diff --git a/internal/twitch/service.go b/internal/twitch/service.go index dcdf3b8..7cd7765 100644 --- a/internal/twitch/service.go +++ b/internal/twitch/service.go @@ -33,6 +33,17 @@ type Channel struct { Login string } +// StreamInfo describes whether a channel is live. +type StreamInfo struct { + Online bool +} + +// PlaybackToken carries the Twitch playback token/signature pair for live HLS access. +type PlaybackToken struct { + Signature string + Value string +} + // StreamerStatus describes one row's resolution state in the UI. type StreamerStatus string @@ -47,6 +58,7 @@ type StreamerEntry struct { ConfigLogin string Login string ChannelID string + Live bool Status StreamerStatus Error string } @@ -106,3 +118,44 @@ func (s *Service) ResolveStreamer(ctx context.Context, login string) (*Channel, } return &Channel{ID: data.User.ID, Login: resolvedLogin}, nil } + +// StreamInfo resolves whether the given channel ID is currently live. +func (s *Service) StreamInfo(ctx context.Context, channelID string) (*StreamInfo, error) { + var data struct { + User *struct { + Stream *struct { + ID string `json:"id"` + } `json:"stream"` + } `json:"user"` + } + if err := s.GQL.Do(ctx, gql.WithIsStreamLiveQuery(channelID), &data); err != nil { + return nil, err + } + if data.User == nil { + return nil, fmt.Errorf("stream info missing user for channel %s", channelID) + } + if data.User.Stream == nil { + return &StreamInfo{Online: false}, nil + } + return &StreamInfo{Online: true}, nil +} + +// PlaybackAccessToken fetches the Twitch playback token needed to access a live stream. +func (s *Service) PlaybackAccessToken(ctx context.Context, login string) (*PlaybackToken, error) { + var data struct { + StreamPlaybackAccessToken *struct { + Signature string `json:"signature"` + Value string `json:"value"` + } `json:"streamPlaybackAccessToken"` + } + if err := s.GQL.Do(ctx, gql.PlaybackAccessToken(login), &data); err != nil { + return nil, err + } + if data.StreamPlaybackAccessToken == nil || data.StreamPlaybackAccessToken.Signature == "" || data.StreamPlaybackAccessToken.Value == "" { + return nil, fmt.Errorf("playback access token missing signature or value for %s", login) + } + return &PlaybackToken{ + Signature: data.StreamPlaybackAccessToken.Signature, + Value: data.StreamPlaybackAccessToken.Value, + }, nil +} diff --git a/internal/twitch/service_test.go b/internal/twitch/service_test.go index 9dd2d21..34873e5 100644 --- a/internal/twitch/service_test.go +++ b/internal/twitch/service_test.go @@ -63,3 +63,64 @@ func TestResolveStreamerNotFound(t *testing.T) { t.Fatalf("error = %v", err) } } + +func TestStreamInfoOffline(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "WithIsStreamLiveQuery": `{"user":{"stream":null}}`, + }} + service := &Service{GQL: client} + info, err := service.StreamInfo(context.Background(), "123") + if err != nil { + t.Fatal(err) + } + if info.Online { + t.Fatalf("info = %#v, want offline", info) + } +} + +func TestStreamInfoOnline(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "WithIsStreamLiveQuery": `{"user":{"stream":{"id":"broadcast"}}}`, + }} + service := &Service{GQL: client} + info, err := service.StreamInfo(context.Background(), "123") + if err != nil { + t.Fatal(err) + } + if !info.Online { + t.Fatalf("info = %#v, want online", info) + } +} + +func TestPlaybackAccessToken(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "PlaybackAccessToken": `{"streamPlaybackAccessToken":{"signature":"sig","value":"token"}}`, + }} + service := &Service{GQL: client} + token, err := service.PlaybackAccessToken(context.Background(), "streamer") + if err != nil { + t.Fatal(err) + } + if token.Signature != "sig" || token.Value != "token" { + t.Fatalf("token = %#v", token) + } +} + +func TestPlaybackAccessTokenMissingFields(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "PlaybackAccessToken": `{"streamPlaybackAccessToken":{"signature":"sig"}}`, + }} + service := &Service{GQL: client} + _, err := service.PlaybackAccessToken(context.Background(), "streamer") + if err == nil { + t.Fatal("expected error") + } +}