diff --git a/internal/gql/operations.go b/internal/gql/operations.go index 01f906b..7221242 100644 --- a/internal/gql/operations.go +++ b/internal/gql/operations.go @@ -95,6 +95,15 @@ func ChannelPointsContext(login string) Request { }) } +// WatchStreak fetches the authenticated viewer's watch-streak milestone for one streamer. +func WatchStreak(login string) Request { + return queryOperation( + "WatchStreak", + `query WatchStreak($login: String!) { user(login: $login) { channel { self { viewerMilestones { id category value } } } } }`, + map[string]any{"login": login}, + ) +} + // ClaimCommunityPoints claims an available channel points bonus chest. func ClaimCommunityPoints(channelID, claimID string) Request { return operation("ClaimCommunityPoints", "46aaeebe02c99afdf4fc97c7c0cba964124bf6b0af229395f1f6d1feed05b3d0", map[string]any{ diff --git a/internal/miner/manager.go b/internal/miner/manager.go index bf6e3ee..3e6e522 100644 --- a/internal/miner/manager.go +++ b/internal/miner/manager.go @@ -11,10 +11,17 @@ import ( ) const defaultWatchInterval = 20 * time.Second +const ( + maxWatchedChannels = 2 + watchStreakMaintenanceLimit = 7 * time.Minute + watchStreakRestartThreshold = 30 * time.Minute + recentOfflineRestartGuard = time.Minute +) // Service is the Twitch capability surface the miner needs. type Service interface { LoadChannelPointsContext(context.Context, string) (*twitch.ChannelPointsContext, error) + WatchStreak(context.Context, string) (*int, error) ClaimCommunityPoints(context.Context, string, string) error StreamMetadata(context.Context, string) (*twitch.StreamMetadata, error) FetchSpadeURL(context.Context, string) (string, error) @@ -35,14 +42,34 @@ type LogEntry struct { Line string } +// WatchReason describes why the miner is currently watching a streamer. +type WatchReason string + +const ( + WatchReasonStreak WatchReason = "watchstreak" + WatchReasonPoints WatchReason = "points" +) + +// StatusEntry is the current miner state for one resolved streamer login. +type StatusEntry struct { + Login string + Watching bool + Reason WatchReason + WatchedMinutes int + WatchStreakMinutes int + WatchStreak *int +} + // Manager owns background channel points mining state for the current authenticated user. type Manager struct { ctx context.Context service Service pubsub PubSubSyncer onLog func(LogEntry) + onStatus func(StatusEntry) watchInterval time.Duration sleep func(context.Context, time.Duration) error + now func() time.Time mu sync.Mutex auth *auth.State @@ -64,11 +91,30 @@ type streamerState struct { Metadata *twitch.StreamMetadata LastWatchAt time.Time + WatchStreak *int + WatchStreakMissing bool + WatchStreakWatched time.Duration + Watched time.Duration + CurrentWatchReason WatchReason + OnlineAt time.Time + OfflineAt time.Time + PendingStreamUpAt time.Time + CurrentBroadcastID string + seeded bool seeding bool refreshing bool } +type watchCandidate struct { + configLogin string + login string + channelID string + spadeURL string + metadata *twitch.StreamMetadata + reason WatchReason +} + // NewManager constructs one background miner manager. func NewManager(ctx context.Context, service Service, pubsub PubSubSyncer, onLog func(LogEntry)) *Manager { manager := &Manager{ @@ -78,6 +124,7 @@ func NewManager(ctx context.Context, service Service, pubsub PubSubSyncer, onLog onLog: onLog, watchInterval: defaultWatchInterval, sleep: sleepContext, + now: time.Now, entries: make(map[string]*streamerState), } if manager.pubsub == nil { @@ -86,6 +133,13 @@ func NewManager(ctx context.Context, service Service, pubsub PubSubSyncer, onLog return manager } +// SetStatusSink wires a separate miner status callback used by the TUI. +func (m *Manager) SetStatusSink(onStatus func(StatusEntry)) { + m.mu.Lock() + m.onStatus = onStatus + m.mu.Unlock() +} + // Sync reconciles the miner against the current resolved streamer snapshot. func (m *Manager) Sync(ctx context.Context, state *auth.State, viewer *twitch.Viewer, entries []twitch.StreamerEntry) { if state == nil || viewer == nil { @@ -99,6 +153,7 @@ func (m *Manager) Sync(ctx context.Context, state *auth.State, viewer *twitch.Vi var ( channelIDs []string seedList []seedTarget + statuses []StatusEntry ) m.mu.Lock() @@ -127,13 +182,25 @@ func (m *Manager) Sync(ctx context.Context, state *auth.State, viewer *twitch.Vi current.ConfigLogin = entry.ConfigLogin current.Login = entry.Login current.ChannelID = entry.ChannelID - current.Live = entry.Live + current.WatchStreak = cloneInt(entry.WatchStreak) + now := m.now() + switch { + case entry.Live && !current.Live: + m.markOnlineConfirmedLocked(current, now, entry.WatchStreak) + statuses = append(statuses, statusFromState(*current)) + case !entry.Live && current.Live: + m.markOfflineLocked(current, now) + statuses = append(statuses, statusFromState(*current)) + default: + current.Live = entry.Live + } nextEntries[entry.ConfigLogin] = current } m.entries = nextEntries m.startWatchLoopLocked() m.mu.Unlock() + m.emitStatuses(statuses) _ = m.pubsub.Sync(ctx, viewer.ID, state.AccessToken, channelIDs) for _, target := range seedList { @@ -245,6 +312,7 @@ func (m *Manager) scheduleRefresh(configLogin string) { if current, ok := m.entries[configLogin]; ok && current.ChannelID == channelID { current.SpadeURL = spadeURL current.Metadata = metadata + current.CurrentBroadcastID = metadata.BroadcastID current.Live = true } m.mu.Unlock() @@ -274,34 +342,12 @@ func (m *Manager) watchLoop(ctx context.Context) { } func (m *Manager) watchOnce(ctx context.Context) error { - type candidate struct { - configLogin string - login string - channelID string - spadeURL string - metadata *twitch.StreamMetadata - } - m.mu.Lock() viewer := m.viewer - candidates := make([]candidate, 0, 2) - for _, configLogin := range m.order { - state, ok := m.entries[configLogin] - if !ok || !state.Live || state.Login == "" || state.ChannelID == "" { - continue - } - candidates = append(candidates, candidate{ - configLogin: configLogin, - login: state.Login, - channelID: state.ChannelID, - spadeURL: state.SpadeURL, - metadata: state.Metadata, - }) - if len(candidates) == 2 { - break - } - } + candidates := m.watchCandidatesLocked() + statuses := m.updateWatchReasonsLocked(candidates) m.mu.Unlock() + m.emitStatuses(statuses) if viewer == nil || len(candidates) == 0 { return nil @@ -330,14 +376,181 @@ func (m *Manager) watchOnce(ctx context.Context) error { } m.mu.Lock() + statuses := []StatusEntry{} if state, ok := m.entries[candidate.configLogin]; ok && state.ChannelID == candidate.channelID { - state.LastWatchAt = time.Now() + state.LastWatchAt = m.now() + state.Watched += time.Minute + if candidate.reason == WatchReasonStreak && state.WatchStreakMissing { + state.WatchStreakWatched += time.Minute + if state.WatchStreakWatched >= watchStreakMaintenanceLimit { + state.WatchStreakMissing = false + state.CurrentWatchReason = WatchReasonPoints + } + } + statuses = append(statuses, statusFromState(*state)) } m.mu.Unlock() + m.emitStatuses(statuses) } return nil } +func (m *Manager) watchCandidatesLocked() []watchCandidate { + selected := make([]watchCandidate, 0, maxWatchedChannels) + seen := make(map[string]bool) + for _, configLogin := range m.order { + if len(selected) == maxWatchedChannels { + break + } + state, ok := m.entries[configLogin] + if !ok || !state.eligibleForStreakWatch() { + continue + } + selected = append(selected, watchCandidateFromState(*state, WatchReasonStreak)) + seen[configLogin] = true + } + + for _, configLogin := range m.order { + if len(selected) == maxWatchedChannels { + break + } + if seen[configLogin] { + continue + } + state, ok := m.entries[configLogin] + if !ok || !state.eligibleForWatch() { + continue + } + selected = append(selected, watchCandidateFromState(*state, WatchReasonPoints)) + seen[configLogin] = true + } + return selected +} + +func watchCandidateFromState(state streamerState, reason WatchReason) watchCandidate { + return watchCandidate{ + configLogin: state.ConfigLogin, + login: state.Login, + channelID: state.ChannelID, + spadeURL: state.SpadeURL, + metadata: state.Metadata, + reason: reason, + } +} + +func (s streamerState) eligibleForWatch() bool { + return s.Live && s.Login != "" && s.ChannelID != "" +} + +func (s streamerState) eligibleForStreakWatch() bool { + if !s.eligibleForWatch() || !s.WatchStreakMissing || s.WatchStreakWatched >= watchStreakMaintenanceLimit { + return false + } + if s.OfflineAt.IsZero() { + return true + } + return !s.OnlineAt.IsZero() && s.OnlineAt.Sub(s.OfflineAt) > watchStreakRestartThreshold +} + +func (m *Manager) updateWatchReasonsLocked(candidates []watchCandidate) []StatusEntry { + reasons := make(map[string]WatchReason, len(candidates)) + for _, candidate := range candidates { + reasons[candidate.configLogin] = candidate.reason + } + + statuses := make([]StatusEntry, 0, len(m.entries)) + for configLogin, state := range m.entries { + nextReason := reasons[configLogin] + if !state.Live { + nextReason = "" + } + if state.CurrentWatchReason == nextReason { + continue + } + state.CurrentWatchReason = nextReason + statuses = append(statuses, statusFromState(*state)) + } + return statuses +} + +func (m *Manager) markOnlineConfirmedLocked(state *streamerState, now time.Time, watchStreak *int) { + state.Live = true + state.PendingStreamUpAt = time.Time{} + if watchStreak != nil { + state.WatchStreak = cloneInt(watchStreak) + } + if !state.OfflineAt.IsZero() && now.Sub(state.OfflineAt) < recentOfflineRestartGuard { + return + } + state.OnlineAt = now + state.WatchStreakMissing = true + state.WatchStreakWatched = 0 + state.SpadeURL = "" + state.Metadata = nil + state.CurrentBroadcastID = "" +} + +func (m *Manager) markOfflineLocked(state *streamerState, now time.Time) { + state.Live = false + state.OfflineAt = now + state.SpadeURL = "" + state.Metadata = nil + state.Watched = 0 + state.CurrentWatchReason = "" + state.PendingStreamUpAt = time.Time{} + state.CurrentBroadcastID = "" +} + +func (m *Manager) emitStatusForConfig(configLogin string) { + m.mu.Lock() + var ( + status StatusEntry + ok bool + ) + if state, found := m.entries[configLogin]; found { + status = statusFromState(*state) + ok = true + } + m.mu.Unlock() + if ok { + m.emitStatuses([]StatusEntry{status}) + } +} + +func (m *Manager) emitStatuses(statuses []StatusEntry) { + if len(statuses) == 0 { + return + } + m.mu.Lock() + onStatus := m.onStatus + m.mu.Unlock() + if onStatus == nil { + return + } + for _, status := range statuses { + onStatus(status) + } +} + +func statusFromState(state streamerState) StatusEntry { + return StatusEntry{ + Login: state.Login, + Watching: state.Live && state.CurrentWatchReason != "", + Reason: state.CurrentWatchReason, + WatchedMinutes: int(state.Watched / time.Minute), + WatchStreakMinutes: int(state.WatchStreakWatched / time.Minute), + WatchStreak: cloneInt(state.WatchStreak), + } +} + +func cloneInt(value *int) *int { + if value == nil { + return nil + } + copy := *value + return © +} + func (m *Manager) handlePubSubEvent(event Event) { configLogin, state := m.streamerForChannel(event.ChannelID) if state == nil { @@ -351,8 +564,18 @@ func (m *Manager) handlePubSubEvent(event Event) { m.mu.Lock() if current, ok := m.entries[configLogin]; ok { current.ChannelPoints = event.Balance + if event.ReasonCode == "WATCH_STREAK" { + current.WatchStreakMissing = false + if current.CurrentWatchReason == WatchReasonStreak { + current.CurrentWatchReason = WatchReasonPoints + } + } } m.mu.Unlock() + if event.ReasonCode == "WATCH_STREAK" { + m.emitStatusForConfig(configLogin) + m.scheduleWatchStreakRefresh(configLogin) + } case "claim-available": if event.ClaimID != "" { m.logf(state.Login, "claiming bonus chest: %s", event.ClaimID) @@ -363,28 +586,66 @@ func (m *Manager) handlePubSubEvent(event Event) { case "stream-up": m.mu.Lock() if current, ok := m.entries[configLogin]; ok { - current.Live = true + current.PendingStreamUpAt = m.now() } m.mu.Unlock() - m.scheduleRefresh(configLogin) case "stream-down": + var statuses []StatusEntry m.mu.Lock() if current, ok := m.entries[configLogin]; ok { - current.Live = false + m.markOfflineLocked(current, m.now()) + statuses = append(statuses, statusFromState(*current)) } m.mu.Unlock() + m.emitStatuses(statuses) case "viewcount": if !state.Live { + var ( + statuses []StatusEntry + shouldRefresh bool + ) m.mu.Lock() if current, ok := m.entries[configLogin]; ok { - current.Live = true + m.markOnlineConfirmedLocked(current, m.now(), nil) + statuses = append(statuses, statusFromState(*current)) + shouldRefresh = current.Live } m.mu.Unlock() - m.scheduleRefresh(configLogin) + m.emitStatuses(statuses) + if shouldRefresh { + m.scheduleRefresh(configLogin) + m.scheduleWatchStreakRefresh(configLogin) + } } } } +func (m *Manager) scheduleWatchStreakRefresh(configLogin string) { + m.mu.Lock() + state, ok := m.entries[configLogin] + if !ok { + m.mu.Unlock() + return + } + login := state.Login + channelID := state.ChannelID + m.mu.Unlock() + + go func() { + streak, err := m.service.WatchStreak(m.ctx, login) + if err != nil { + m.logf(login, "watch streak refresh failed: %v", err) + return + } + m.mu.Lock() + if current, ok := m.entries[configLogin]; ok && current.ChannelID == channelID { + current.WatchStreak = cloneInt(streak) + } + m.mu.Unlock() + m.emitStatusForConfig(configLogin) + }() +} + func (m *Manager) logEvent(login string, event Event) { switch event.MessageType { case "points-earned": diff --git a/internal/miner/manager_test.go b/internal/miner/manager_test.go index f6a6be4..53f17e7 100644 --- a/internal/miner/manager_test.go +++ b/internal/miner/manager_test.go @@ -15,7 +15,9 @@ type fakeService struct { channelPoints map[string]*twitch.ChannelPointsContext metadata map[string]*twitch.StreamMetadata spadeURLs map[string]string + watchStreaks map[string]int playbackErr map[string]error + minuteErr map[string]error claimErr map[string]error claimed []string @@ -30,6 +32,16 @@ func (f *fakeService) LoadChannelPointsContext(_ context.Context, login string) return nil, errors.New("missing channel points") } +func (f *fakeService) WatchStreak(_ context.Context, login string) (*int, error) { + if f.watchStreaks != nil { + if value, ok := f.watchStreaks[login]; ok { + return &value, nil + } + } + value := 0 + return &value, nil +} + func (f *fakeService) ClaimCommunityPoints(_ context.Context, channelID, claimID string) error { f.claimed = append(f.claimed, channelID+":"+claimID) if err, ok := f.claimErr[channelID+":"+claimID]; ok { @@ -67,6 +79,9 @@ func (f *fakeService) TouchPlayback(_ context.Context, login string, _ *twitch.P func (f *fakeService) SendMinuteWatched(_ context.Context, spadeURL string, _ twitch.MinuteWatchedPayload) error { f.watched = append(f.watched, "post:"+spadeURL) + if err, ok := f.minuteErr[spadeURL]; ok { + return err + } return nil } @@ -158,6 +173,287 @@ func TestManagerWatchOnceUsesTopTwoLiveStreamers(t *testing.T) { } } +func TestManagerWatchOncePrioritizesMissingWatchStreaks(t *testing.T) { + service := readyWatchService("alpha_live", "beta_live", "gamma_live") + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.viewer = &twitch.Viewer{ID: "viewer"} + manager.order = []string{"alpha", "beta", "gamma"} + manager.entries = map[string]*streamerState{ + "alpha": readyState("alpha", "alpha_live", "1", false), + "beta": readyState("beta", "beta_live", "2", true), + "gamma": readyState("gamma", "gamma_live", "3", true), + } + + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + + if got, want := service.watched, []string{ + "touch:beta_live", + "post:https://spade.test/beta_live", + "touch:gamma_live", + "post:https://spade.test/gamma_live", + }; !reflect.DeepEqual(got, want) { + t.Fatalf("watched = %#v, want %#v", got, want) + } +} + +func TestManagerWatchOnceFillsWithPointsCandidates(t *testing.T) { + service := readyWatchService("alpha_live", "beta_live", "gamma_live") + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.viewer = &twitch.Viewer{ID: "viewer"} + manager.order = []string{"alpha", "beta", "gamma"} + manager.entries = map[string]*streamerState{ + "alpha": readyState("alpha", "alpha_live", "1", false), + "beta": readyState("beta", "beta_live", "2", true), + "gamma": readyState("gamma", "gamma_live", "3", false), + } + + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + + if got, want := service.watched, []string{ + "touch:beta_live", + "post:https://spade.test/beta_live", + "touch:alpha_live", + "post:https://spade.test/alpha_live", + }; !reflect.DeepEqual(got, want) { + t.Fatalf("watched = %#v, want %#v", got, want) + } + if manager.entries["beta"].CurrentWatchReason != WatchReasonStreak { + t.Fatalf("beta reason = %q, want streak", manager.entries["beta"].CurrentWatchReason) + } + if manager.entries["alpha"].CurrentWatchReason != WatchReasonPoints { + t.Fatalf("alpha reason = %q, want points", manager.entries["alpha"].CurrentWatchReason) + } +} + +func TestManagerStopsWatchStreakMaintenanceAfterSevenMinutes(t *testing.T) { + service := readyWatchService("alpha_live") + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.viewer = &twitch.Viewer{ID: "viewer"} + manager.order = []string{"alpha"} + manager.entries = map[string]*streamerState{ + "alpha": readyState("alpha", "alpha_live", "1", true), + } + + for i := 0; i < 7; i++ { + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + } + + state := manager.entries["alpha"] + if state.WatchStreakMissing { + t.Fatal("expected watch streak maintenance to stop") + } + if state.WatchStreakWatched != watchStreakMaintenanceLimit { + t.Fatalf("watched streak duration = %s, want %s", state.WatchStreakWatched, watchStreakMaintenanceLimit) + } + if state.Watched != watchStreakMaintenanceLimit { + t.Fatalf("watched duration = %s, want %s", state.Watched, watchStreakMaintenanceLimit) + } + if state.CurrentWatchReason != WatchReasonPoints { + t.Fatalf("watch reason = %q, want points", state.CurrentWatchReason) + } +} + +func TestManagerTracksSuccessfulWatchedMinutesAndResetsOffline(t *testing.T) { + service := readyWatchService("alpha_live") + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.viewer = &twitch.Viewer{ID: "viewer"} + manager.order = []string{"alpha"} + manager.entries = map[string]*streamerState{ + "alpha": readyState("alpha", "alpha_live", "1", false), + } + + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + + state := manager.entries["alpha"] + if state.Watched != 2*time.Minute { + t.Fatalf("watched duration = %s, want 2m", state.Watched) + } + status := statusFromState(*state) + if status.WatchedMinutes != 2 { + t.Fatalf("status watched minutes = %d, want 2", status.WatchedMinutes) + } + + manager.handlePubSubEvent(Event{MessageType: "stream-down", ChannelID: "1", Timestamp: "down", Topic: "video-playback-by-id"}) + + state = manager.entries["alpha"] + if state.Watched != 0 { + t.Fatalf("watched duration after offline = %s, want 0", state.Watched) + } + status = statusFromState(*state) + if status.Watching || status.WatchedMinutes != 0 { + t.Fatalf("status after offline = %#v, want not watching with 0 watched minutes", status) + } +} + +func TestManagerDoesNotTrackWatchedMinutesWhenTelemetryFails(t *testing.T) { + service := readyWatchService("alpha_live") + service.minuteErr = map[string]error{ + "https://spade.test/alpha_live": errors.New("minute rejected"), + } + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.viewer = &twitch.Viewer{ID: "viewer"} + manager.order = []string{"alpha"} + manager.entries = map[string]*streamerState{ + "alpha": readyState("alpha", "alpha_live", "1", false), + } + + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + + if state := manager.entries["alpha"]; state.Watched != 0 { + t.Fatalf("watched duration = %s, want 0 after failed telemetry", state.Watched) + } +} + +func TestHandlePubSubWatchStreakCompletesMaintenance(t *testing.T) { + service := &fakeService{watchStreaks: map[string]int{"alpha_live": 8}} + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.entries["alpha"] = &streamerState{ + ConfigLogin: "alpha", + ChannelID: "1", + Login: "alpha_live", + Live: true, + WatchStreakMissing: true, + CurrentWatchReason: WatchReasonStreak, + } + + manager.handlePubSubEvent(Event{MessageType: "points-earned", ChannelID: "1", Balance: 555, ReasonCode: "WATCH_STREAK", TotalPoints: 450, Timestamp: "t1", Topic: "community-points-user-v1"}) + + waitFor(t, func() bool { + manager.mu.Lock() + defer manager.mu.Unlock() + state := manager.entries["alpha"] + return !state.WatchStreakMissing && state.CurrentWatchReason == WatchReasonPoints && state.WatchStreak != nil && *state.WatchStreak == 8 + }) +} + +func TestManagerLifecycleResetsWatchStreakOnConfirmedOnline(t *testing.T) { + service := &fakeService{} + now := time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC) + manager := NewManager(context.Background(), service, &fakePubSub{}, nil) + defer manager.Close() + manager.now = func() time.Time { return now } + manager.entries["alpha"] = &streamerState{ + ConfigLogin: "alpha", + ChannelID: "1", + Login: "alpha_live", + Live: true, + WatchStreakMissing: false, + WatchStreakWatched: 5 * time.Minute, + SpadeURL: "https://spade.test/alpha", + Metadata: &twitch.StreamMetadata{BroadcastID: "old"}, + } + + manager.handlePubSubEvent(Event{MessageType: "stream-down", ChannelID: "1", Timestamp: "down", Topic: "video-playback-by-id"}) + now = now.Add(31 * time.Minute) + manager.handlePubSubEvent(Event{MessageType: "viewcount", ChannelID: "1", Timestamp: "viewcount", Topic: "video-playback-by-id"}) + + manager.mu.Lock() + defer manager.mu.Unlock() + state := manager.entries["alpha"] + if !state.Live || !state.WatchStreakMissing || state.WatchStreakWatched != 0 { + t.Fatalf("state = %#v, want fresh live streak maintenance", state) + } + if state.SpadeURL != "" || state.Metadata != nil { + t.Fatalf("playback metadata was not cleared: %#v", state) + } +} + +func TestManagerLifecycleRecentOfflineGuardPreservesStreakState(t *testing.T) { + now := time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC) + manager := NewManager(context.Background(), &fakeService{}, &fakePubSub{}, nil) + defer manager.Close() + manager.now = func() time.Time { return now } + manager.entries["alpha"] = &streamerState{ + ConfigLogin: "alpha", + ChannelID: "1", + Login: "alpha_live", + Live: false, + OfflineAt: now.Add(-30 * time.Second), + WatchStreakMissing: false, + WatchStreakWatched: 5 * time.Minute, + } + + manager.handlePubSubEvent(Event{MessageType: "viewcount", ChannelID: "1", Timestamp: "viewcount", Topic: "video-playback-by-id"}) + + manager.mu.Lock() + defer manager.mu.Unlock() + state := manager.entries["alpha"] + if !state.Live { + t.Fatal("expected channel to be live after viewcount") + } + if state.WatchStreakMissing || state.WatchStreakWatched != 5*time.Minute { + t.Fatalf("state = %#v, want recent offline guard to preserve streak state", state) + } +} + +func TestManagerSyncPollingConfirmedOnlineResetsStreakState(t *testing.T) { + now := time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC) + manager := NewManager(context.Background(), &fakeService{}, &fakePubSub{}, nil) + defer manager.Close() + manager.sleep = cancelSleep + manager.now = func() time.Time { return now } + manager.entries["alpha"] = &streamerState{ + ConfigLogin: "alpha", + ChannelID: "1", + Login: "alpha_live", + Live: false, + OfflineAt: now.Add(-31 * time.Minute), + WatchStreakMissing: false, + WatchStreakWatched: 5 * time.Minute, + } + streak := 9 + + manager.Sync(context.Background(), &auth.State{AccessToken: "token"}, &twitch.Viewer{ID: "viewer"}, []twitch.StreamerEntry{ + {ConfigLogin: "alpha", Login: "alpha_live", ChannelID: "1", Live: true, Status: twitch.StreamerReady, WatchStreak: &streak}, + }) + + manager.mu.Lock() + defer manager.mu.Unlock() + state := manager.entries["alpha"] + if !state.Live || !state.WatchStreakMissing || state.WatchStreakWatched != 0 || state.WatchStreak == nil || *state.WatchStreak != 9 { + t.Fatalf("state = %#v, want polling-confirmed fresh live streak state", state) + } +} + +func TestManagerStreamUpWaitsForConfirmation(t *testing.T) { + now := time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC) + manager := NewManager(context.Background(), &fakeService{}, &fakePubSub{}, nil) + defer manager.Close() + manager.now = func() time.Time { return now } + manager.entries["alpha"] = &streamerState{ConfigLogin: "alpha", ChannelID: "1", Login: "alpha_live"} + + manager.handlePubSubEvent(Event{MessageType: "stream-up", ChannelID: "1", Timestamp: "up", Topic: "video-playback-by-id"}) + + manager.mu.Lock() + defer manager.mu.Unlock() + state := manager.entries["alpha"] + if state.Live { + t.Fatal("stream-up should wait for API/viewcount confirmation") + } + if !state.PendingStreamUpAt.Equal(now) { + t.Fatalf("pending stream-up = %s, want %s", state.PendingStreamUpAt, now) + } +} + func TestHandlePubSubEventUpdatesBalanceAndClaims(t *testing.T) { service := &fakeService{} manager := NewManager(context.Background(), service, &fakePubSub{}, nil) @@ -205,6 +501,31 @@ func TestManagerLogsPubSubEventsAndFailures(t *testing.T) { func cancelSleep(context.Context, time.Duration) error { return context.Canceled } +func readyWatchService(logins ...string) *fakeService { + service := &fakeService{ + metadata: map[string]*twitch.StreamMetadata{}, + spadeURLs: map[string]string{}, + } + for _, login := range logins { + service.metadata[login] = &twitch.StreamMetadata{BroadcastID: "broadcast-" + login} + service.spadeURLs[login] = "https://spade.test/" + login + } + return service +} + +func readyState(configLogin, login, channelID string, missingStreak bool) *streamerState { + return &streamerState{ + ConfigLogin: configLogin, + Login: login, + ChannelID: channelID, + Live: true, + SpadeURL: "https://spade.test/" + login, + Metadata: &twitch.StreamMetadata{BroadcastID: "broadcast-" + login}, + WatchStreakMissing: missingStreak, + OnlineAt: time.Date(2026, 4, 30, 12, 0, 0, 0, time.UTC), + } +} + func waitFor(t *testing.T, condition func() bool) { t.Helper() deadline := time.Now().Add(2 * time.Second) diff --git a/internal/miner/pubsub.go b/internal/miner/pubsub.go index e222f77..a96b898 100644 --- a/internal/miner/pubsub.go +++ b/internal/miner/pubsub.go @@ -28,6 +28,8 @@ type Event struct { Timestamp string Balance int ClaimID string + ReasonCode string + TotalPoints int } func (e Event) key() string { @@ -291,6 +293,10 @@ func parseFrame(message []byte) (*frame, error) { Balance int `json:"balance"` ChannelID string `json:"channel_id"` } `json:"balance"` + PointGain *struct { + ReasonCode string `json:"reason_code"` + TotalPoints int `json:"total_points"` + } `json:"point_gain"` } `json:"data"` } if err := json.Unmarshal([]byte(envelope.Data.Message), &payload); err != nil { @@ -320,6 +326,10 @@ func parseFrame(message []byte) (*frame, error) { if payload.Data.Claim != nil { result.Event.ClaimID = payload.Data.Claim.ID } + if payload.Data.PointGain != nil { + result.Event.ReasonCode = payload.Data.PointGain.ReasonCode + result.Event.TotalPoints = payload.Data.PointGain.TotalPoints + } return result, nil } diff --git a/internal/miner/pubsub_test.go b/internal/miner/pubsub_test.go index a831e32..ad78f8d 100644 --- a/internal/miner/pubsub_test.go +++ b/internal/miner/pubsub_test.go @@ -14,6 +14,18 @@ func TestParseFramePointsEarned(t *testing.T) { } } +func TestParseFramePointsEarnedPointGain(t *testing.T) { + t.Parallel() + + frame, err := parseFrame([]byte(`{"type":"MESSAGE","data":{"topic":"community-points-user-v1.viewer","message":"{\"type\":\"points-earned\",\"data\":{\"timestamp\":\"2026-04-29T12:00:00Z\",\"balance\":{\"balance\":42,\"channel_id\":\"123\"},\"point_gain\":{\"reason_code\":\"WATCH_STREAK\",\"total_points\":450}}}"}}`)) + if err != nil { + t.Fatal(err) + } + if frame.Event.ReasonCode != "WATCH_STREAK" || frame.Event.TotalPoints != 450 { + t.Fatalf("event = %#v", frame.Event) + } +} + func TestParseFrameClaimAvailable(t *testing.T) { t.Parallel() diff --git a/internal/tui/messages.go b/internal/tui/messages.go index 5e5fd1c..228f284 100644 --- a/internal/tui/messages.go +++ b/internal/tui/messages.go @@ -42,8 +42,18 @@ type IRCUpdate struct { // MinerUpdate carries one miner log line into the TUI. type MinerUpdate struct { - Login string - Line string + Login string + Line string + Status *MinerStatus +} + +// MinerStatus carries the current active watch state for one streamer. +type MinerStatus struct { + Watching bool + Reason string + WatchedMinutes int + WatchStreakMinutes int + WatchStreak *int } type authStartedMsg struct { diff --git a/internal/tui/model.go b/internal/tui/model.go index 72c56c4..091f107 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -61,6 +61,7 @@ type Model struct { focus panelFocus ircDetails map[string]ircDetail minerDetails map[string][]string + minerStatuses map[string]MinerStatus authViewport viewport.Model ircViewport viewport.Model minerViewport viewport.Model @@ -79,18 +80,19 @@ func New(options Options) Model { } model := Model{ - streamers: streamers, - authState: options.AuthState, - runtime: options.runtime, - authLogs: []string{}, - mode: authView, - ircDetails: make(map[string]ircDetail), - minerDetails: make(map[string][]string), - width: defaultViewWidth, - height: 24, - focus: focusStreamers, - authViewport: newAuthViewport(contentWidth(defaultViewWidth), authViewportHeight(24)), - ircViewport: newIRCViewport(detailViewportWidth(defaultViewWidth), detailViewportHeight(24, "")), + streamers: streamers, + authState: options.AuthState, + runtime: options.runtime, + authLogs: []string{}, + mode: authView, + ircDetails: make(map[string]ircDetail), + minerDetails: make(map[string][]string), + minerStatuses: make(map[string]MinerStatus), + width: defaultViewWidth, + height: 24, + focus: focusStreamers, + authViewport: newAuthViewport(contentWidth(defaultViewWidth), authViewportHeight(24)), + ircViewport: newIRCViewport(detailViewportWidth(defaultViewWidth), detailViewportHeight(24, "")), minerViewport: newMinerViewport( detailViewportWidth(defaultViewWidth), detailViewportHeight(24, ""), @@ -204,6 +206,7 @@ func (m Model) updateAuth(msg AuthUpdate) (tea.Model, tea.Cmd) { m.streamers = loadingEntries(m.streamers) m.ircDetails = make(map[string]ircDetail) m.minerDetails = make(map[string][]string) + m.minerStatuses = make(map[string]MinerStatus) m.selectedConfig = "" m.focus = focusStreamers m.ensureSelection() @@ -269,11 +272,16 @@ func (m *Model) applyIRCUpdate(update IRCUpdate) { func (m *Model) applyMinerUpdate(update MinerUpdate) { login := normalizeKey(update.Login) - line := strings.TrimSpace(update.Line) - if login == "" || line == "" { + if login == "" { return } - m.minerDetails[login] = appendCappedHistory(m.minerDetails[login], line) + if update.Status != nil { + m.minerStatuses[login] = *update.Status + } + line := strings.TrimSpace(update.Line) + if line != "" { + m.minerDetails[login] = appendCappedHistory(m.minerDetails[login], line) + } } func (m *Model) ensureSelection() { diff --git a/internal/tui/model_test.go b/internal/tui/model_test.go index 230e399..28521c3 100644 --- a/internal/tui/model_test.go +++ b/internal/tui/model_test.go @@ -128,6 +128,141 @@ func TestViewDisplaysInactiveDetailForOfflineStreamer(t *testing.T) { assertContainsAll(t, model.View(), "offline", "inactive") } +func TestInfoTabShowsFetchingWatchStreak(t *testing.T) { + model := dashboardModel(twitch.StreamerEntry{ConfigLogin: "alpha", Login: "alpha_live", Status: twitch.StreamerReady}) + entry, ok := model.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + assertLastInfoLine(t, model.renderInfoTab(entry), "Watch streak: fetching") +} + +func TestInfoTabShowsFetchedWatchStreak(t *testing.T) { + streak := 3 + model := dashboardModel(twitch.StreamerEntry{ + ConfigLogin: "alpha", + Login: "alpha_live", + Live: true, + Status: twitch.StreamerReady, + WatchStreak: &streak, + }) + entry, ok := model.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + assertLastInfoLine(t, model.renderInfoTab(entry), "Watch streak: 3") +} + +func TestInfoTabShowsUnavailableWatchStreak(t *testing.T) { + model := dashboardModel(twitch.StreamerEntry{ + ConfigLogin: "alpha", + Login: "alpha_live", + Live: true, + Status: twitch.StreamerReady, + WatchStreakError: "lookup failed", + }) + entry, ok := model.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + assertLastInfoLine(t, model.renderInfoTab(entry), "Watch streak: unavailable") +} + +func TestInfoTabShowsWatchStreakMinerStatus(t *testing.T) { + streak := 3 + model := dashboardModel(twitch.StreamerEntry{ + ConfigLogin: "alpha", + Login: "alpha_live", + Live: true, + Status: twitch.StreamerReady, + WatchStreak: &streak, + }) + updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ + Login: "alpha_live", + Status: &MinerStatus{ + Watching: true, + Reason: "watchstreak", + WatchedMinutes: 4, + }, + }}) + next := updated.(Model) + entry, ok := next.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + info := next.renderInfoTab(entry) + assertContainsAll(t, info, "Watching for watchstreak", "Watched for 4 minutes so far") + assertNotContains(t, info, "Watching for watchstreak (4 mins so far)") + assertLastInfoLine(t, info, "Watched for 4 minutes so far") +} + +func TestInfoTabShowsPointsMinerStatus(t *testing.T) { + streak := 3 + minerStreak := 4 + model := dashboardModel(twitch.StreamerEntry{ + ConfigLogin: "alpha", + Login: "alpha_live", + Live: true, + Status: twitch.StreamerReady, + WatchStreak: &streak, + }) + updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ + Login: "alpha_live", + Status: &MinerStatus{ + Watching: true, + Reason: "points", + WatchedMinutes: 6, + WatchStreak: &minerStreak, + }, + }}) + next := updated.(Model) + entry, ok := next.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + info := next.renderInfoTab(entry) + assertContainsAll(t, info, "Watch streak: 4", "Watching for points", "Watched for 6 minutes so far") + assertLastInfoLine(t, info, "Watched for 6 minutes so far") +} + +func TestInfoTabOmitsMinerStatusWhenInactiveOrUnwatched(t *testing.T) { + streak := 3 + model := dashboardModel(twitch.StreamerEntry{ + ConfigLogin: "alpha", + Login: "alpha_live", + Status: twitch.StreamerReady, + WatchStreak: &streak, + }) + updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ + Login: "alpha_live", + Status: &MinerStatus{ + Watching: true, + Reason: "points", + }, + }}) + next := updated.(Model) + entry, ok := next.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + assertNotContains(t, next.renderInfoTab(entry), "Watching for") + assertNotContains(t, next.renderInfoTab(entry), "Watched for") + + next = dashboardModel(twitch.StreamerEntry{ + ConfigLogin: "alpha", + Login: "alpha_live", + Live: true, + Status: twitch.StreamerReady, + WatchStreak: &streak, + }) + entry, ok = next.selectedEntry() + if !ok { + t.Fatal("expected selected entry") + } + assertNotContains(t, next.renderInfoTab(entry), "Watching for") + assertNotContains(t, next.renderInfoTab(entry), "Watched for") +} + func assertInOrder(t *testing.T, value string, parts ...string) { t.Helper() @@ -589,3 +724,18 @@ func assertContainsAll(t *testing.T, got string, wants ...string) { } } } + +func assertNotContains(t *testing.T, got string, want string) { + t.Helper() + if strings.Contains(got, want) { + t.Fatalf("unexpected %q in:\n%s", want, got) + } +} + +func assertLastInfoLine(t *testing.T, got, want string) { + t.Helper() + lines := strings.Split(got, "\n") + if len(lines) == 0 || lines[len(lines)-1] != want { + t.Fatalf("last line = %q in:\n%s", lines[len(lines)-1], got) + } +} diff --git a/internal/tui/runtime.go b/internal/tui/runtime.go index 81d1a66..252d00e 100644 --- a/internal/tui/runtime.go +++ b/internal/tui/runtime.go @@ -156,6 +156,7 @@ type streamerService interface { CurrentUser(context.Context) (*twitch.Viewer, error) ResolveStreamer(context.Context, string) (*twitch.Channel, error) StreamInfo(context.Context, string) (*twitch.StreamInfo, error) + WatchStreak(context.Context, string) (*int, error) PlaybackAccessToken(context.Context, string) (*twitch.PlaybackToken, error) LoadChannelPointsContext(context.Context, string) (*twitch.ChannelPointsContext, error) ClaimCommunityPoints(context.Context, string, string) error @@ -209,7 +210,9 @@ func newIRCManager(ch chan<- StreamerUpdate) *irc.Manager { } func newMinerManager(ctx context.Context, service miner.Service, ch chan<- StreamerUpdate) *miner.Manager { - return miner.NewManager(ctx, service, nil, newMinerLogSink(ch)) + manager := miner.NewManager(ctx, service, nil, newMinerLogSink(ch)) + manager.SetStatusSink(newMinerStatusSink(ch)) + return manager } func newMinerLogSink(ch chan<- StreamerUpdate) func(miner.LogEntry) { @@ -223,6 +226,23 @@ func newMinerLogSink(ch chan<- StreamerUpdate) func(miner.LogEntry) { } } +func newMinerStatusSink(ch chan<- StreamerUpdate) func(miner.StatusEntry) { + return func(entry miner.StatusEntry) { + ch <- StreamerUpdate{ + Miner: &MinerUpdate{ + Login: entry.Login, + Status: &MinerStatus{ + Watching: entry.Watching, + Reason: string(entry.Reason), + WatchedMinutes: entry.WatchedMinutes, + WatchStreakMinutes: entry.WatchStreakMinutes, + WatchStreak: cloneInt(entry.WatchStreak), + }, + }, + } + } +} + func resolveStreamerEntries(ctx context.Context, service streamerService, state *auth.State, logins []string, ircSyncer ircSyncer, minerSyncer minerSyncer, send func(StreamerUpdate)) error { return resolveStreamerEntriesWithSleep(ctx, service, state, logins, ircSyncer, minerSyncer, send, streamerRefreshInterval, sleepContext) } @@ -275,6 +295,16 @@ func resolveStreamerEntry(ctx context.Context, service streamerService, login st return entry, nil } + watchStreak, err := service.WatchStreak(ctx, channel.Login) + switch { + case err == nil: + entry.WatchStreak = watchStreak + case errors.Is(err, context.Canceled): + return nil, err + default: + entry.WatchStreakError = err.Error() + } + stream, err := service.StreamInfo(ctx, channel.ID) switch { case err == nil: @@ -331,3 +361,11 @@ func watchedIRCTargets(entries []twitch.StreamerEntry) []irc.Target { } return targets } + +func cloneInt(value *int) *int { + if value == nil { + return nil + } + copy := *value + return © +} diff --git a/internal/tui/runtime_test.go b/internal/tui/runtime_test.go index 979078e..dc870d6 100644 --- a/internal/tui/runtime_test.go +++ b/internal/tui/runtime_test.go @@ -18,6 +18,7 @@ type fakeStreamerService struct { resolveErrs map[string]error streams map[string][]streamResult streamCalls map[string]int + streaks map[string]watchStreakResult playbackErr map[string]error } @@ -53,6 +54,18 @@ func (f *fakeStreamerService) StreamInfo(_ context.Context, channelID string) (* return results[call].info, nil } +func (f *fakeStreamerService) WatchStreak(_ context.Context, login string) (*int, error) { + result, ok := f.streaks[login] + if !ok { + value := 0 + return &value, nil + } + if result.err != nil { + return nil, result.err + } + return &result.value, nil +} + func (f *fakeStreamerService) PlaybackAccessToken(_ context.Context, login string) (*twitch.PlaybackToken, error) { if err, ok := f.playbackErr[login]; ok { return nil, err @@ -145,6 +158,35 @@ func TestResolveStreamerEntriesPlaybackFailureStillMarksLive(t *testing.T) { } } +func TestResolveStreamerEntriesWatchStreakFailureStillMarksLive(t *testing.T) { + updates := collectUpdates(t, streamService( + channel("alpha", "1", "alpha_live"), + live("1", true), + watchStreakErr("alpha_live", errors.New("watch streak failed")), + ), []string{"alpha"}, nil, nil, cancelAfter(0)) + + entry := updates[1].Entry + if entry == nil || entry.Status != twitch.StreamerReady || !entry.Live { + t.Fatalf("entry = %#v", entry) + } + if entry.WatchStreakError == "" { + t.Fatalf("entry = %#v, want watch streak error", entry) + } +} + +func TestResolveStreamerEntriesStoresWatchStreak(t *testing.T) { + updates := collectUpdates(t, streamService( + channel("alpha", "1", "alpha_live"), + live("1", false), + watchStreak("alpha_live", 9), + ), []string{"alpha"}, nil, nil, cancelAfter(0)) + + entry := updates[1].Entry + if entry == nil || entry.WatchStreak == nil || *entry.WatchStreak != 9 { + t.Fatalf("entry = %#v, want watch streak 9", entry) + } +} + func TestResolveStreamerEntriesRefreshesLiveState(t *testing.T) { updates := collectUpdates(t, streamService( channel("alpha", "1", "alpha_live"), @@ -226,12 +268,18 @@ func TestNewMinerLogSinkBridgesMinerEntriesIntoStreamerUpdates(t *testing.T) { type serviceOption func(*fakeStreamerService) +type watchStreakResult struct { + value int + err error +} + func streamService(options ...serviceOption) *fakeStreamerService { service := &fakeStreamerService{ channels: map[string]*twitch.Channel{}, resolveErrs: map[string]error{}, streams: map[string][]streamResult{}, streamCalls: map[string]int{}, + streaks: map[string]watchStreakResult{}, playbackErr: map[string]error{}, } for _, option := range options { @@ -270,6 +318,18 @@ func playbackErr(login string, err error) serviceOption { } } +func watchStreak(login string, value int) serviceOption { + return func(service *fakeStreamerService) { + service.streaks[login] = watchStreakResult{value: value} + } +} + +func watchStreakErr(login string, err error) serviceOption { + return func(service *fakeStreamerService) { + service.streaks[login] = watchStreakResult{err: err} + } +} + func collectUpdates(t *testing.T, service *fakeStreamerService, logins []string, ircSyncer ircSyncer, minerSyncer minerSyncer, sleep func(context.Context, time.Duration) error) []StreamerUpdate { t.Helper() var updates []StreamerUpdate diff --git a/internal/tui/view.go b/internal/tui/view.go index 6a06abf..e8c3337 100644 --- a/internal/tui/view.go +++ b/internal/tui/view.go @@ -128,6 +128,8 @@ func (m Model) renderInfoTab(entry twitch.StreamerEntry) string { detail := m.ircDetails[normalizeKey(entry.Login)] if !isActive(entry) { lines = append(lines, "Chat: "+mutedStyle.Render("inactive")) + lines = append(lines, m.watchStreakText(entry)) + lines = append(lines, m.minerWatchingLines(entry)...) return strings.Join(lines, "\n") } @@ -136,6 +138,8 @@ func (m Model) renderInfoTab(entry twitch.StreamerEntry) string { } else { lines = append(lines, "Chat: "+mutedStyle.Render("not joined")) } + lines = append(lines, m.watchStreakText(entry)) + lines = append(lines, m.minerWatchingLines(entry)...) return strings.Join(lines, "\n") } @@ -220,6 +224,50 @@ func statusText(entry twitch.StreamerEntry) string { } } +func (m Model) watchStreakText(entry twitch.StreamerEntry) string { + if status, ok := m.minerStatuses[normalizeKey(entry.Login)]; ok && status.WatchStreak != nil { + return fmt.Sprintf("Watch streak: %d", *status.WatchStreak) + } + switch { + case entry.WatchStreak != nil: + return fmt.Sprintf("Watch streak: %d", *entry.WatchStreak) + case entry.WatchStreakError != "": + return "Watch streak: unavailable" + default: + return "Watch streak: fetching" + } +} + +func (m Model) minerWatchingLines(entry twitch.StreamerEntry) []string { + if !isActive(entry) { + return nil + } + status, ok := m.minerStatuses[normalizeKey(entry.Login)] + if !ok || !status.Watching { + return nil + } + var reason string + switch status.Reason { + case string(minerWatchReasonStreak): + reason = "Watching for watchstreak" + case string(minerWatchReasonPoints): + reason = "Watching for points" + default: + return nil + } + return []string{ + reason, + fmt.Sprintf("Watched for %d minutes so far", status.WatchedMinutes), + } +} + +type minerWatchReason string + +const ( + minerWatchReasonStreak minerWatchReason = "watchstreak" + minerWatchReasonPoints minerWatchReason = "points" +) + func rawStatus(entry twitch.StreamerEntry) string { switch { case entry.Status == twitch.StreamerError: diff --git a/internal/twitch/service.go b/internal/twitch/service.go index acce753..9e57f5b 100644 --- a/internal/twitch/service.go +++ b/internal/twitch/service.go @@ -5,14 +5,15 @@ package twitch import ( "bytes" + "context" "encoding/base64" "encoding/json" - "context" "errors" "fmt" "io" "net/http" "regexp" + "strconv" "strings" "parasocial/internal/gql" @@ -25,7 +26,7 @@ const ( ) var ( - defaultPageURL = "https://www.twitch.tv/" + defaultPageURL = "https://www.twitch.tv/" settingsAssetPattern = regexp.MustCompile(spadeSettingsRegex) spadeURLPattern = regexp.MustCompile(spadeURLRegex) ) @@ -124,12 +125,14 @@ const ( // StreamerEntry is the UI-facing state for one configured streamer. type StreamerEntry struct { - ConfigLogin string - Login string - ChannelID string - Live bool - Status StreamerStatus - Error string + ConfigLogin string + Login string + ChannelID string + Live bool + Status StreamerStatus + Error string + WatchStreak *int + WatchStreakError string } // ErrStreamerNotFound is returned when Twitch has no channel for the requested login. @@ -258,6 +261,41 @@ func (s *Service) LoadChannelPointsContext(ctx context.Context, login string) (* return context, nil } +// WatchStreak fetches the authenticated viewer's current watch streak for one streamer. +func (s *Service) WatchStreak(ctx context.Context, login string) (*int, error) { + var data struct { + User *struct { + Channel *struct { + Self *struct { + ViewerMilestones []struct { + ID string `json:"id"` + Category string `json:"category"` + Value string `json:"value"` + } `json:"viewerMilestones"` + } `json:"self"` + } `json:"channel"` + } `json:"user"` + } + if err := s.GQL.Do(ctx, gql.WatchStreak(login), &data); err != nil { + return nil, err + } + if data.User == nil || data.User.Channel == nil || data.User.Channel.Self == nil { + return nil, fmt.Errorf("watch streak missing viewer milestones for %s", login) + } + for _, milestone := range data.User.Channel.Self.ViewerMilestones { + if milestone.Category != "WATCH_STREAK" { + continue + } + value, err := strconv.Atoi(strings.TrimSpace(milestone.Value)) + if err != nil { + return nil, fmt.Errorf("parse watch streak for %s: %w", login, err) + } + return &value, nil + } + value := 0 + return &value, nil +} + // ClaimCommunityPoints claims one available channel points bonus chest. func (s *Service) ClaimCommunityPoints(ctx context.Context, channelID, claimID string) error { var data struct { diff --git a/internal/twitch/service_test.go b/internal/twitch/service_test.go index 254049b..04b06cc 100644 --- a/internal/twitch/service_test.go +++ b/internal/twitch/service_test.go @@ -146,6 +146,51 @@ func TestLoadChannelPointsContext(t *testing.T) { } } +func TestWatchStreakParsesMilestoneValue(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "WatchStreak": `{"user":{"channel":{"self":{"viewerMilestones":[{"id":"1","category":"WATCH_STREAK","value":"12"}]}}}}`, + }} + service := &Service{GQL: client} + streak, err := service.WatchStreak(context.Background(), "streamer") + if err != nil { + t.Fatal(err) + } + if streak == nil || *streak != 12 { + t.Fatalf("streak = %#v, want 12", streak) + } +} + +func TestWatchStreakReturnsZeroWhenMilestoneMissing(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "WatchStreak": `{"user":{"channel":{"self":{"viewerMilestones":[{"id":"1","category":"OTHER","value":"12"}]}}}}`, + }} + service := &Service{GQL: client} + streak, err := service.WatchStreak(context.Background(), "streamer") + if err != nil { + t.Fatal(err) + } + if streak == nil || *streak != 0 { + t.Fatalf("streak = %#v, want 0", streak) + } +} + +func TestWatchStreakMalformedValueErrors(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "WatchStreak": `{"user":{"channel":{"self":{"viewerMilestones":[{"id":"1","category":"WATCH_STREAK","value":"many"}]}}}}`, + }} + service := &Service{GQL: client} + _, err := service.WatchStreak(context.Background(), "streamer") + if err == nil { + t.Fatal("expected error") + } +} + func TestClaimCommunityPoints(t *testing.T) { t.Parallel()