From b631d201b01a1ec24344911772f94e8d03c3e909 Mon Sep 17 00:00:00 2001 From: ruinivist Date: Wed, 29 Apr 2026 22:33:55 +0200 Subject: [PATCH] feat: add twitch channel points miner runtime --- go.mod | 1 + go.sum | 2 + internal/gql/client_test.go | 15 ++ internal/gql/operations.go | 39 ++++ internal/miner/manager.go | 384 ++++++++++++++++++++++++++++++++ internal/miner/manager_test.go | 188 ++++++++++++++++ internal/miner/pubsub.go | 354 +++++++++++++++++++++++++++++ internal/miner/pubsub_test.go | 39 ++++ internal/tui/runtime.go | 42 +++- internal/tui/runtime_test.go | 61 ++++- internal/twitch/service.go | 356 ++++++++++++++++++++++++++++- internal/twitch/service_test.go | 178 +++++++++++++++ 12 files changed, 1640 insertions(+), 19 deletions(-) create mode 100644 internal/miner/manager.go create mode 100644 internal/miner/manager_test.go create mode 100644 internal/miner/pubsub.go create mode 100644 internal/miner/pubsub_test.go diff --git a/go.mod b/go.mod index c113853..2f2c5f6 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/clipperhouse/stringish v0.1.1 // indirect github.com/clipperhouse/uax29/v2 v2.5.0 // indirect github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect + github.com/gorilla/websocket v1.5.3 // indirect github.com/lucasb-eyer/go-colorful v1.3.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-localereader v0.0.1 // indirect diff --git a/go.sum b/go.sum index 7c992c1..8592b4f 100644 --- a/go.sum +++ b/go.sum @@ -24,6 +24,8 @@ github.com/clipperhouse/uax29/v2 v2.5.0 h1:x7T0T4eTHDONxFJsL94uKNKPHrclyFI0lm7+w github.com/clipperhouse/uax29/v2 v2.5.0/go.mod h1:Wn1g7MK6OoeDT0vL+Q0SQLDz/KpfsVRgg6W7ihQeh4g= github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f h1:Y/CXytFA4m6baUTXGLOoWe4PQhGxaX0KpnayAqC48p4= github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f/go.mod h1:vw97MGsxSvLiUE2X8qFplwetxpGLQrlU1Q9AUEIzCaM= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag= github.com/lucasb-eyer/go-colorful v1.3.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= diff --git a/internal/gql/client_test.go b/internal/gql/client_test.go index 566cbdb..455667a 100644 --- a/internal/gql/client_test.go +++ b/internal/gql/client_test.go @@ -141,4 +141,19 @@ func TestPersistedOperationHashes(t *testing.T) { if got := req.Extensions.PersistedQuery.SHA256Hash; got != "94e82a7b1e3c21e186daa73ee2afc4b8f23bade1fbbff6fe8ac133f50a2f58ca" { t.Fatalf("hash = %q", got) } + + channelPoints := ChannelPointsContext("streamer") + if got := channelPoints.Extensions.PersistedQuery.SHA256Hash; got != "1530a003a7d374b0380b79db0be0534f30ff46e61cffa2bc0e2468a909fbc024" { + t.Fatalf("channel points hash = %q", got) + } + + claim := ClaimCommunityPoints("7", "claim-1") + if got := claim.Extensions.PersistedQuery.SHA256Hash; got != "46aaeebe02c99afdf4fc97c7c0cba964124bf6b0af229395f1f6d1feed05b3d0" { + t.Fatalf("claim hash = %q", got) + } + + streamInfo := VideoPlayerStreamInfoOverlayChannel("streamer") + if got := streamInfo.Extensions.PersistedQuery.SHA256Hash; got != "198492e0857f6aedead9665c81c5a06d67b25b58034649687124083ff288597d" { + t.Fatalf("stream info hash = %q", got) + } } diff --git a/internal/gql/operations.go b/internal/gql/operations.go index 0bef505..01f906b 100644 --- a/internal/gql/operations.go +++ b/internal/gql/operations.go @@ -3,6 +3,8 @@ // and streamer login resolution without exposing raw payload assembly to callers. package gql +import "strings" + // persistedQuery stores the persisted-query metadata Twitch expects. type persistedQuery struct { Version int `json:"version"` @@ -85,3 +87,40 @@ func WithIsStreamLiveQuery(channelID string) Request { "id": channelID, }) } + +// ChannelPointsContext fetches channel points state for one streamer login. +func ChannelPointsContext(login string) Request { + return operation("ChannelPointsContext", "1530a003a7d374b0380b79db0be0534f30ff46e61cffa2bc0e2468a909fbc024", map[string]any{ + "channelLogin": login, + }) +} + +// ClaimCommunityPoints claims an available channel points bonus chest. +func ClaimCommunityPoints(channelID, claimID string) Request { + return operation("ClaimCommunityPoints", "46aaeebe02c99afdf4fc97c7c0cba964124bf6b0af229395f1f6d1feed05b3d0", map[string]any{ + "input": map[string]any{ + "channelID": channelID, + "claimID": claimID, + }, + }) +} + +// VideoPlayerStreamInfoOverlayChannel fetches stream metadata used for watch telemetry. +func VideoPlayerStreamInfoOverlayChannel(login string) Request { + return operation("VideoPlayerStreamInfoOverlayChannel", "198492e0857f6aedead9665c81c5a06d67b25b58034649687124083ff288597d", map[string]any{ + "channel": login, + }) +} + +// HLSMasterPlaylistURL builds the usher URL Twitch uses for live playback. +func HLSMasterPlaylistURL(login, signature, token string) string { + var builder strings.Builder + builder.Grow(len(login) + len(signature) + len(token) + 80) + builder.WriteString("https://usher.ttvnw.net/api/channel/hls/") + builder.WriteString(login) + builder.WriteString(".m3u8?player=twitchweb&allow_source=true&type=any&p=1&sig=") + builder.WriteString(signature) + builder.WriteString("&token=") + builder.WriteString(token) + return builder.String() +} diff --git a/internal/miner/manager.go b/internal/miner/manager.go new file mode 100644 index 0000000..9583196 --- /dev/null +++ b/internal/miner/manager.go @@ -0,0 +1,384 @@ +package miner + +import ( + "context" + "sync" + "time" + + "parasocial/internal/auth" + "parasocial/internal/twitch" +) + +const defaultWatchInterval = 20 * time.Second + +// Service is the Twitch capability surface the miner needs. +type Service interface { + LoadChannelPointsContext(context.Context, string) (*twitch.ChannelPointsContext, error) + ClaimCommunityPoints(context.Context, string, string) error + StreamMetadata(context.Context, string) (*twitch.StreamMetadata, error) + FetchSpadeURL(context.Context, string) (string, error) + PlaybackAccessToken(context.Context, string) (*twitch.PlaybackToken, error) + TouchPlayback(context.Context, string, *twitch.PlaybackToken) error + SendMinuteWatched(context.Context, string, twitch.MinuteWatchedPayload) error +} + +// PubSubSyncer reconciles the live PubSub subscriptions the miner needs. +type PubSubSyncer interface { + Sync(context.Context, string, string, []string) error + Close() error +} + +// Manager owns background channel points mining state for the current authenticated user. +type Manager struct { + ctx context.Context + service Service + pubsub PubSubSyncer + watchInterval time.Duration + sleep func(context.Context, time.Duration) error + + mu sync.Mutex + auth *auth.State + viewer *twitch.Viewer + order []string + entries map[string]*streamerState + + watchStarted bool + watchCancel context.CancelFunc +} + +type streamerState struct { + ConfigLogin string + Login string + ChannelID string + Live bool + ChannelPoints int + SpadeURL string + Metadata *twitch.StreamMetadata + LastWatchAt time.Time + + seeded bool + seeding bool + refreshing bool +} + +// NewManager constructs one background miner manager. +func NewManager(ctx context.Context, service Service, pubsub PubSubSyncer) *Manager { + manager := &Manager{ + ctx: ctx, + service: service, + pubsub: pubsub, + watchInterval: defaultWatchInterval, + sleep: sleepContext, + entries: make(map[string]*streamerState), + } + if manager.pubsub == nil { + manager.pubsub = NewPubSubClient(ctx, manager.handlePubSubEvent) + } + return manager +} + +// 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 { + return + } + + type seedTarget struct { + configLogin string + } + + var ( + channelIDs []string + seedList []seedTarget + ) + + m.mu.Lock() + m.auth = state + m.viewer = viewer + m.order = m.order[:0] + + nextEntries := make(map[string]*streamerState, len(entries)) + for _, entry := range entries { + m.order = append(m.order, entry.ConfigLogin) + if entry.Status != twitch.StreamerReady || entry.ChannelID == "" || entry.Login == "" { + continue + } + + channelIDs = append(channelIDs, entry.ChannelID) + current, ok := m.entries[entry.ConfigLogin] + if !ok || current.ChannelID != entry.ChannelID || current.Login != entry.Login { + current = &streamerState{ + ConfigLogin: entry.ConfigLogin, + Login: entry.Login, + ChannelID: entry.ChannelID, + } + seedList = append(seedList, seedTarget{configLogin: entry.ConfigLogin}) + } + + current.ConfigLogin = entry.ConfigLogin + current.Login = entry.Login + current.ChannelID = entry.ChannelID + current.Live = entry.Live + nextEntries[entry.ConfigLogin] = current + } + + m.entries = nextEntries + m.startWatchLoopLocked() + m.mu.Unlock() + + _ = m.pubsub.Sync(ctx, viewer.ID, state.AccessToken, channelIDs) + for _, target := range seedList { + m.scheduleSeed(target.configLogin) + } +} + +// Close stops background work and closes the PubSub connection. +func (m *Manager) Close() { + m.mu.Lock() + cancel := m.watchCancel + m.watchCancel = nil + m.mu.Unlock() + + if cancel != nil { + cancel() + } + if m.pubsub != nil { + _ = m.pubsub.Close() + } +} + +func (m *Manager) startWatchLoopLocked() { + if m.watchStarted { + return + } + ctx, cancel := context.WithCancel(m.ctx) + m.watchStarted = true + m.watchCancel = cancel + go m.watchLoop(ctx) +} + +func (m *Manager) scheduleSeed(configLogin string) { + m.mu.Lock() + state, ok := m.entries[configLogin] + if !ok || state.seeding { + m.mu.Unlock() + return + } + state.seeding = true + login := state.Login + channelID := state.ChannelID + live := state.Live + m.mu.Unlock() + + go func() { + defer m.finishSeeding(configLogin) + + channelPoints, err := m.service.LoadChannelPointsContext(m.ctx, login) + if err == nil { + m.mu.Lock() + if current, ok := m.entries[configLogin]; ok && current.ChannelID == channelID { + current.ChannelPoints = channelPoints.Balance + current.seeded = true + } + m.mu.Unlock() + if channelPoints.ClaimID != "" { + _ = m.service.ClaimCommunityPoints(m.ctx, channelID, channelPoints.ClaimID) + } + } + + if live { + m.scheduleRefresh(configLogin) + } + }() +} + +func (m *Manager) finishSeeding(configLogin string) { + m.mu.Lock() + defer m.mu.Unlock() + if state, ok := m.entries[configLogin]; ok { + state.seeding = false + } +} + +func (m *Manager) scheduleRefresh(configLogin string) { + m.mu.Lock() + state, ok := m.entries[configLogin] + if !ok || state.refreshing { + m.mu.Unlock() + return + } + state.refreshing = true + login := state.Login + channelID := state.ChannelID + m.mu.Unlock() + + go func() { + defer m.finishRefresh(configLogin) + + spadeURL, err := m.service.FetchSpadeURL(m.ctx, login) + if err != nil { + return + } + metadata, err := m.service.StreamMetadata(m.ctx, login) + if err != nil { + return + } + + m.mu.Lock() + defer m.mu.Unlock() + if current, ok := m.entries[configLogin]; ok && current.ChannelID == channelID { + current.SpadeURL = spadeURL + current.Metadata = metadata + current.Live = true + } + }() +} + +func (m *Manager) finishRefresh(configLogin string) { + m.mu.Lock() + defer m.mu.Unlock() + if state, ok := m.entries[configLogin]; ok { + state.refreshing = false + } +} + +func (m *Manager) watchLoop(ctx context.Context) { + for { + if err := m.watchOnce(ctx); err != nil { + if err == context.Canceled { + return + } + } + if err := m.sleep(ctx, m.watchInterval); err != nil { + return + } + } +} + +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 + } + } + m.mu.Unlock() + + if viewer == nil || len(candidates) == 0 { + return nil + } + + for _, candidate := range candidates { + if candidate.spadeURL == "" || candidate.metadata == nil { + m.scheduleRefresh(candidate.configLogin) + continue + } + + token, err := m.service.PlaybackAccessToken(ctx, candidate.login) + if err != nil { + continue + } + if err := m.service.TouchPlayback(ctx, candidate.login, token); err != nil { + continue + } + + payload := twitch.BuildMinuteWatchedPayload(viewer.ID, candidate.channelID, candidate.login, candidate.metadata) + if err := m.service.SendMinuteWatched(ctx, candidate.spadeURL, payload); err != nil { + continue + } + + m.mu.Lock() + if state, ok := m.entries[candidate.configLogin]; ok && state.ChannelID == candidate.channelID { + state.LastWatchAt = time.Now() + } + m.mu.Unlock() + } + return nil +} + +func (m *Manager) handlePubSubEvent(event Event) { + configLogin, state := m.streamerForChannel(event.ChannelID) + if state == nil { + return + } + + switch event.MessageType { + case "points-earned", "points-spent": + m.mu.Lock() + if current, ok := m.entries[configLogin]; ok { + current.ChannelPoints = event.Balance + } + m.mu.Unlock() + case "claim-available": + if event.ClaimID != "" { + _ = m.service.ClaimCommunityPoints(m.ctx, state.ChannelID, event.ClaimID) + } + case "stream-up": + m.mu.Lock() + if current, ok := m.entries[configLogin]; ok { + current.Live = true + } + m.mu.Unlock() + m.scheduleRefresh(configLogin) + case "stream-down": + m.mu.Lock() + if current, ok := m.entries[configLogin]; ok { + current.Live = false + } + m.mu.Unlock() + case "viewcount": + if !state.Live { + m.mu.Lock() + if current, ok := m.entries[configLogin]; ok { + current.Live = true + } + m.mu.Unlock() + m.scheduleRefresh(configLogin) + } + } +} + +func (m *Manager) streamerForChannel(channelID string) (string, *streamerState) { + m.mu.Lock() + defer m.mu.Unlock() + for configLogin, state := range m.entries { + if state.ChannelID == channelID { + copy := *state + return configLogin, © + } + } + return "", 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/miner/manager_test.go b/internal/miner/manager_test.go new file mode 100644 index 0000000..4e13d5f --- /dev/null +++ b/internal/miner/manager_test.go @@ -0,0 +1,188 @@ +package miner + +import ( + "context" + "errors" + "reflect" + "testing" + "time" + + "parasocial/internal/auth" + "parasocial/internal/twitch" +) + +type fakeService struct { + channelPoints map[string]*twitch.ChannelPointsContext + metadata map[string]*twitch.StreamMetadata + spadeURLs map[string]string + playbackErr map[string]error + + claimed []string + watched []string + refreshed []string +} + +func (f *fakeService) LoadChannelPointsContext(_ context.Context, login string) (*twitch.ChannelPointsContext, error) { + if result, ok := f.channelPoints[login]; ok { + return result, nil + } + return nil, errors.New("missing channel points") +} + +func (f *fakeService) ClaimCommunityPoints(_ context.Context, channelID, claimID string) error { + f.claimed = append(f.claimed, channelID+":"+claimID) + return nil +} + +func (f *fakeService) StreamMetadata(_ context.Context, login string) (*twitch.StreamMetadata, error) { + f.refreshed = append(f.refreshed, login) + if result, ok := f.metadata[login]; ok { + return result, nil + } + return nil, errors.New("missing metadata") +} + +func (f *fakeService) FetchSpadeURL(_ context.Context, login string) (string, error) { + if result, ok := f.spadeURLs[login]; ok { + return result, nil + } + return "", errors.New("missing spade url") +} + +func (f *fakeService) 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 +} + +func (f *fakeService) TouchPlayback(_ context.Context, login string, _ *twitch.PlaybackToken) error { + f.watched = append(f.watched, "touch:"+login) + return nil +} + +func (f *fakeService) SendMinuteWatched(_ context.Context, spadeURL string, _ twitch.MinuteWatchedPayload) error { + f.watched = append(f.watched, "post:"+spadeURL) + return nil +} + +type fakePubSub struct { + syncCalls [][]string +} + +func (f *fakePubSub) Sync(_ context.Context, _, _ string, channelIDs []string) error { + f.syncCalls = append(f.syncCalls, append([]string(nil), channelIDs...)) + return nil +} + +func (f *fakePubSub) Close() error { return nil } + +func TestManagerSyncSeedsAndClaims(t *testing.T) { + service := &fakeService{ + channelPoints: map[string]*twitch.ChannelPointsContext{ + "alpha_live": {Balance: 250, ClaimID: "claim-1"}, + }, + metadata: map[string]*twitch.StreamMetadata{ + "alpha_live": {BroadcastID: "broadcast"}, + }, + spadeURLs: map[string]string{ + "alpha_live": "https://spade.test/alpha", + }, + } + pubsub := &fakePubSub{} + manager := NewManager(context.Background(), service, pubsub) + defer manager.Close() + manager.sleep = cancelSleep + + 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}, + }) + + waitFor(t, func() bool { + manager.mu.Lock() + defer manager.mu.Unlock() + return len(service.claimed) == 1 && manager.entries["alpha"].ChannelPoints == 250 && manager.entries["alpha"].SpadeURL != "" + }) + + if got, want := pubsub.syncCalls, [][]string{{"1"}}; !reflect.DeepEqual(got, want) { + t.Fatalf("syncCalls = %#v, want %#v", got, want) + } +} + +func TestManagerWatchOnceUsesTopTwoLiveStreamers(t *testing.T) { + service := &fakeService{ + metadata: map[string]*twitch.StreamMetadata{ + "alpha_live": {BroadcastID: "a"}, + "beta_live": {BroadcastID: "b"}, + }, + spadeURLs: map[string]string{ + "alpha_live": "https://spade.test/alpha", + "beta_live": "https://spade.test/beta", + }, + channelPoints: map[string]*twitch.ChannelPointsContext{ + "alpha_live": {Balance: 10}, + "beta_live": {Balance: 20}, + "gamma_live": {Balance: 30}, + }, + } + manager := NewManager(context.Background(), service, &fakePubSub{}) + defer manager.Close() + manager.sleep = cancelSleep + + 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}, + {ConfigLogin: "beta", Login: "beta_live", ChannelID: "2", Live: true, Status: twitch.StreamerReady}, + {ConfigLogin: "gamma", Login: "gamma_live", ChannelID: "3", Live: true, Status: twitch.StreamerReady}, + }) + + waitFor(t, func() bool { + manager.mu.Lock() + defer manager.mu.Unlock() + return manager.entries["alpha"].SpadeURL != "" && manager.entries["beta"].SpadeURL != "" + }) + + if err := manager.watchOnce(context.Background()); err != nil { + t.Fatal(err) + } + if got, want := service.watched, []string{ + "touch:alpha_live", + "post:https://spade.test/alpha", + "touch:beta_live", + "post:https://spade.test/beta", + }; !reflect.DeepEqual(got, want) { + t.Fatalf("watched = %#v, want %#v", got, want) + } +} + +func TestHandlePubSubEventUpdatesBalanceAndClaims(t *testing.T) { + service := &fakeService{} + manager := NewManager(context.Background(), service, &fakePubSub{}) + defer manager.Close() + manager.entries["alpha"] = &streamerState{ConfigLogin: "alpha", ChannelID: "1", Login: "alpha_live", Live: true} + + manager.handlePubSubEvent(Event{MessageType: "points-earned", ChannelID: "1", Balance: 555, Timestamp: "t1", Topic: "community-points-user-v1"}) + manager.handlePubSubEvent(Event{MessageType: "claim-available", ChannelID: "1", ClaimID: "claim-2", Timestamp: "t2", Topic: "community-points-user-v1"}) + + manager.mu.Lock() + defer manager.mu.Unlock() + if manager.entries["alpha"].ChannelPoints != 555 { + t.Fatalf("channelPoints = %d", manager.entries["alpha"].ChannelPoints) + } + if got, want := service.claimed, []string{"1:claim-2"}; !reflect.DeepEqual(got, want) { + t.Fatalf("claimed = %#v, want %#v", got, want) + } +} + +func cancelSleep(context.Context, time.Duration) error { return context.Canceled } + +func waitFor(t *testing.T, condition func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if condition() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("condition not met before timeout") +} diff --git a/internal/miner/pubsub.go b/internal/miner/pubsub.go new file mode 100644 index 0000000..e222f77 --- /dev/null +++ b/internal/miner/pubsub.go @@ -0,0 +1,354 @@ +package miner + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "sort" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +const ( + defaultPubSubURL = "wss://pubsub-edge.twitch.tv/v1" + pingInterval = 25 * time.Second + stalePongDeadline = 55 * time.Second + reconnectDelay = 5 * time.Second +) + +// Event is one parsed Twitch PubSub message relevant to the miner. +type Event struct { + Topic string + MessageType string + ChannelID string + Timestamp string + Balance int + ClaimID string +} + +func (e Event) key() string { + return fmt.Sprintf("%s|%s|%s|%s", e.MessageType, e.Topic, e.ChannelID, e.Timestamp) +} + +// Client maintains Twitch PubSub subscriptions and dispatches decoded miner events. +type Client struct { + ctx context.Context + cancel context.CancelFunc + onEvent func(Event) + url string + dialer *websocket.Dialer + mu sync.Mutex + conn *websocket.Conn + viewerID string + token string + topics []string + updateCh chan struct{} + lastPong time.Time + lastSeen string +} + +// NewPubSubClient constructs a PubSub client with the default Twitch endpoint. +func NewPubSubClient(ctx context.Context, onEvent func(Event)) *Client { + runCtx, cancel := context.WithCancel(ctx) + client := &Client{ + ctx: runCtx, + cancel: cancel, + onEvent: onEvent, + url: defaultPubSubURL, + dialer: websocket.DefaultDialer, + updateCh: make(chan struct{}, 1), + lastPong: time.Now(), + } + go client.run() + return client +} + +// Sync updates the desired user and channel topic set. +func (c *Client) Sync(_ context.Context, viewerID, accessToken string, channelIDs []string) error { + topics := make([]string, 0, len(channelIDs)+1) + if viewerID != "" { + topics = append(topics, "community-points-user-v1."+viewerID) + } + for _, channelID := range channelIDs { + if channelID == "" { + continue + } + topics = append(topics, "video-playback-by-id."+channelID) + } + sort.Strings(topics) + + c.mu.Lock() + c.viewerID = viewerID + c.token = accessToken + changed := !equalStrings(c.topics, topics) + c.topics = topics + conn := c.conn + c.mu.Unlock() + + if changed && conn != nil { + _ = conn.Close() + } + select { + case c.updateCh <- struct{}{}: + default: + } + return nil +} + +// Close stops the PubSub run loop and closes any active socket. +func (c *Client) Close() error { + c.cancel() + c.mu.Lock() + conn := c.conn + c.conn = nil + c.mu.Unlock() + if conn != nil { + return conn.Close() + } + return nil +} + +func (c *Client) run() { + for { + topics, token := c.snapshot() + if len(topics) == 0 || token == "" { + select { + case <-c.ctx.Done(): + return + case <-c.updateCh: + continue + } + } + + if err := c.runConnection(topics, token); err != nil && errors.Is(err, context.Canceled) { + return + } + + if err := sleepContext(c.ctx, reconnectDelay); err != nil { + return + } + } +} + +func (c *Client) runConnection(topics []string, token string) error { + conn, _, err := c.dialer.DialContext(c.ctx, c.url, http.Header{ + "User-Agent": []string{"parasocial"}, + }) + if err != nil { + return err + } + defer conn.Close() + + c.mu.Lock() + c.conn = conn + c.lastPong = time.Now() + c.lastSeen = "" + c.mu.Unlock() + defer func() { + c.mu.Lock() + if c.conn == conn { + c.conn = nil + } + c.mu.Unlock() + }() + + if err := c.listenTopics(conn, topics, token); err != nil { + return err + } + + pingErr := make(chan error, 1) + go c.pingLoop(conn, pingErr) + + for { + select { + case <-c.ctx.Done(): + return c.ctx.Err() + case err := <-pingErr: + return err + default: + } + + _, message, err := conn.ReadMessage() + if err != nil { + return err + } + frame, err := parseFrame(message) + if err != nil { + continue + } + switch frame.Type { + case "PONG": + c.mu.Lock() + c.lastPong = time.Now() + c.mu.Unlock() + case "RECONNECT": + return nil + case "MESSAGE": + if frame.Event.key() == "" { + continue + } + c.mu.Lock() + duplicate := c.lastSeen == frame.Event.key() + if !duplicate { + c.lastSeen = frame.Event.key() + } + c.mu.Unlock() + if !duplicate && c.onEvent != nil { + c.onEvent(frame.Event) + } + } + } +} + +func (c *Client) pingLoop(conn *websocket.Conn, errCh chan<- error) { + ticker := time.NewTicker(pingInterval) + defer ticker.Stop() + for { + select { + case <-c.ctx.Done(): + return + case <-ticker.C: + if err := conn.WriteJSON(map[string]any{"type": "PING"}); err != nil { + select { + case errCh <- err: + default: + } + return + } + c.mu.Lock() + stale := time.Since(c.lastPong) > stalePongDeadline + c.mu.Unlock() + if stale { + select { + case errCh <- errors.New("stale pong from twitch pubsub"): + default: + } + return + } + } + } +} + +func (c *Client) listenTopics(conn *websocket.Conn, topics []string, token string) error { + payload := map[string]any{ + "type": "LISTEN", + "nonce": fmt.Sprintf("%d", time.Now().UnixNano()), + "data": map[string]any{ + "topics": topics, + }, + } + if token != "" { + payload["data"].(map[string]any)["auth_token"] = token + } + return conn.WriteJSON(payload) +} + +func (c *Client) snapshot() ([]string, string) { + c.mu.Lock() + defer c.mu.Unlock() + topics := append([]string(nil), c.topics...) + return topics, c.token +} + +type frame struct { + Type string + Error string + Event Event +} + +func parseFrame(message []byte) (*frame, error) { + var envelope struct { + Type string `json:"type"` + Error string `json:"error"` + Data *struct { + Topic string `json:"topic"` + Message string `json:"message"` + } `json:"data"` + } + if err := json.Unmarshal(message, &envelope); err != nil { + return nil, err + } + + result := &frame{Type: envelope.Type, Error: envelope.Error} + if envelope.Type != "MESSAGE" || envelope.Data == nil { + return result, nil + } + + var payload struct { + Type string `json:"type"` + Data struct { + Timestamp string `json:"timestamp"` + ChannelID string `json:"channel_id"` + Claim *struct { + ID string `json:"id"` + ChannelID string `json:"channel_id"` + } `json:"claim"` + Balance *struct { + Balance int `json:"balance"` + ChannelID string `json:"channel_id"` + } `json:"balance"` + } `json:"data"` + } + if err := json.Unmarshal([]byte(envelope.Data.Message), &payload); err != nil { + return nil, err + } + + channelID := payload.Data.ChannelID + if payload.Data.Claim != nil && payload.Data.Claim.ChannelID != "" { + channelID = payload.Data.Claim.ChannelID + } + if payload.Data.Balance != nil && payload.Data.Balance.ChannelID != "" { + channelID = payload.Data.Balance.ChannelID + } + if channelID == "" { + channelID = topicSuffix(envelope.Data.Topic) + } + + result.Event = Event{ + Topic: topicPrefix(envelope.Data.Topic), + MessageType: payload.Type, + ChannelID: channelID, + Timestamp: payload.Data.Timestamp, + } + if payload.Data.Balance != nil { + result.Event.Balance = payload.Data.Balance.Balance + } + if payload.Data.Claim != nil { + result.Event.ClaimID = payload.Data.Claim.ID + } + return result, nil +} + +func topicPrefix(topic string) string { + for i := len(topic) - 1; i >= 0; i-- { + if topic[i] == '.' { + return topic[:i] + } + } + return topic +} + +func topicSuffix(topic string) string { + for i := len(topic) - 1; i >= 0; i-- { + if topic[i] == '.' { + return topic[i+1:] + } + } + return "" +} + +func equalStrings(left, right []string) bool { + if len(left) != len(right) { + return false + } + for i := range left { + if left[i] != right[i] { + return false + } + } + return true +} diff --git a/internal/miner/pubsub_test.go b/internal/miner/pubsub_test.go new file mode 100644 index 0000000..a831e32 --- /dev/null +++ b/internal/miner/pubsub_test.go @@ -0,0 +1,39 @@ +package miner + +import "testing" + +func TestParseFramePointsEarned(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\"}}}"}}`)) + if err != nil { + t.Fatal(err) + } + if frame.Event.Topic != "community-points-user-v1" || frame.Event.MessageType != "points-earned" || frame.Event.ChannelID != "123" || frame.Event.Balance != 42 { + t.Fatalf("event = %#v", frame.Event) + } +} + +func TestParseFrameClaimAvailable(t *testing.T) { + t.Parallel() + + frame, err := parseFrame([]byte(`{"type":"MESSAGE","data":{"topic":"community-points-user-v1.viewer","message":"{\"type\":\"claim-available\",\"data\":{\"timestamp\":\"2026-04-29T12:00:00Z\",\"claim\":{\"id\":\"claim-1\",\"channel_id\":\"123\"}}}"}}`)) + if err != nil { + t.Fatal(err) + } + if frame.Event.ClaimID != "claim-1" || frame.Event.ChannelID != "123" { + t.Fatalf("event = %#v", frame.Event) + } +} + +func TestParseFrameVideoPlaybackFallsBackToTopicSuffix(t *testing.T) { + t.Parallel() + + frame, err := parseFrame([]byte(`{"type":"MESSAGE","data":{"topic":"video-playback-by-id.321","message":"{\"type\":\"stream-down\",\"data\":{\"timestamp\":\"2026-04-29T12:00:00Z\"}}"}}`)) + if err != nil { + t.Fatal(err) + } + if frame.Event.Topic != "video-playback-by-id" || frame.Event.ChannelID != "321" || frame.Event.MessageType != "stream-down" { + t.Fatalf("event = %#v", frame.Event) + } +} diff --git a/internal/tui/runtime.go b/internal/tui/runtime.go index caef95a..998b7a3 100644 --- a/internal/tui/runtime.go +++ b/internal/tui/runtime.go @@ -15,6 +15,7 @@ import ( "parasocial/internal/auth" "parasocial/internal/gql" "parasocial/internal/irc" + "parasocial/internal/miner" "parasocial/internal/twitch" ) @@ -135,7 +136,10 @@ func (r *runtime) startResolve(state *auth.State, ch chan<- StreamerUpdate) { ch <- StreamerUpdate{Err: err, Done: true} return } - if err := resolveStreamerEntriesWithSleep(r.ctx, service, state, r.logins, ircManager, func(update StreamerUpdate) { + minerManager := miner.NewManager(r.ctx, service, nil) + defer minerManager.Close() + + if err := resolveStreamerEntriesWithSleep(r.ctx, service, state, r.logins, ircManager, minerManager, func(update StreamerUpdate) { ch <- update }, r.refreshEvery, r.sleep); err != nil { if errors.Is(err, context.Canceled) { @@ -153,12 +157,22 @@ type streamerService interface { ResolveStreamer(context.Context, string) (*twitch.Channel, error) StreamInfo(context.Context, string) (*twitch.StreamInfo, error) PlaybackAccessToken(context.Context, string) (*twitch.PlaybackToken, error) + LoadChannelPointsContext(context.Context, string) (*twitch.ChannelPointsContext, error) + ClaimCommunityPoints(context.Context, string, string) error + StreamMetadata(context.Context, string) (*twitch.StreamMetadata, error) + FetchSpadeURL(context.Context, string) (string, error) + TouchPlayback(context.Context, string, *twitch.PlaybackToken) error + SendMinuteWatched(context.Context, string, twitch.MinuteWatchedPayload) error } type ircSyncer interface { Sync(context.Context, string, string, []irc.Target) } +type minerSyncer interface { + Sync(context.Context, *auth.State, *twitch.Viewer, []twitch.StreamerEntry) +} + func newTwitchService(httpClient *http.Client, state *auth.State) (*twitch.Service, error) { client := &gql.Client{ HTTPClient: httpClient, @@ -172,7 +186,11 @@ func newTwitchService(httpClient *http.Client, state *auth.State) (*twitch.Servi if err := client.Validate(); err != nil { return nil, fmt.Errorf("configure graphql session: %w", err) } - return &twitch.Service{GQL: client}, nil + return &twitch.Service{ + GQL: client, + HTTPClient: httpClient, + Session: client.Session, + }, nil } func newIRCManager(ch chan<- StreamerUpdate) *irc.Manager { @@ -190,11 +208,11 @@ func newIRCManager(ch chan<- StreamerUpdate) *irc.Manager { } } -func resolveStreamerEntries(ctx context.Context, service streamerService, state *auth.State, logins []string, syncer ircSyncer, send func(StreamerUpdate)) error { - return resolveStreamerEntriesWithSleep(ctx, service, state, logins, syncer, send, streamerRefreshInterval, sleepContext) +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) } -func resolveStreamerEntriesWithSleep(ctx context.Context, service streamerService, state *auth.State, logins []string, syncer ircSyncer, send func(StreamerUpdate), interval time.Duration, sleep func(context.Context, time.Duration) error) error { +func resolveStreamerEntriesWithSleep(ctx context.Context, service streamerService, state *auth.State, logins []string, ircSyncer ircSyncer, minerSyncer minerSyncer, send func(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) @@ -202,7 +220,7 @@ func resolveStreamerEntriesWithSleep(ctx context.Context, service streamerServic send(StreamerUpdate{Viewer: viewer}) entries := twitch.LoadingStreamerEntries(logins) - syncWatchedChannels(ctx, syncer, state, entries) + syncRuntimeState(ctx, ircSyncer, minerSyncer, state, viewer, entries) for { for index, login := range logins { @@ -215,7 +233,7 @@ func resolveStreamerEntriesWithSleep(ctx context.Context, service streamerServic Index: index, Entry: entry, }) - syncWatchedChannels(ctx, syncer, state, entries) + syncRuntimeState(ctx, ircSyncer, minerSyncer, state, viewer, entries) } if err := sleep(ctx, interval); err != nil { return err @@ -276,11 +294,13 @@ func sleepContext(ctx context.Context, duration time.Duration) error { } } -func syncWatchedChannels(ctx context.Context, syncer ircSyncer, state *auth.State, entries []twitch.StreamerEntry) { - if syncer == nil || state == nil { - return +func syncRuntimeState(ctx context.Context, ircSyncer ircSyncer, minerSyncer minerSyncer, state *auth.State, viewer *twitch.Viewer, entries []twitch.StreamerEntry) { + if ircSyncer != nil && state != nil { + ircSyncer.Sync(ctx, state.Login, state.AccessToken, watchedIRCTargets(entries)) + } + if minerSyncer != nil { + minerSyncer.Sync(ctx, state, viewer, entries) } - syncer.Sync(ctx, state.Login, state.AccessToken, watchedIRCTargets(entries)) } func watchedIRCTargets(entries []twitch.StreamerEntry) []irc.Target { diff --git a/internal/tui/runtime_test.go b/internal/tui/runtime_test.go index 529542b..b17bc68 100644 --- a/internal/tui/runtime_test.go +++ b/internal/tui/runtime_test.go @@ -59,6 +59,30 @@ func (f *fakeStreamerService) PlaybackAccessToken(_ context.Context, login strin return &twitch.PlaybackToken{Signature: "sig", Value: "token"}, nil } +func (f *fakeStreamerService) LoadChannelPointsContext(context.Context, string) (*twitch.ChannelPointsContext, error) { + return &twitch.ChannelPointsContext{Balance: 0}, nil +} + +func (f *fakeStreamerService) ClaimCommunityPoints(context.Context, string, string) error { + return nil +} + +func (f *fakeStreamerService) StreamMetadata(context.Context, string) (*twitch.StreamMetadata, error) { + return &twitch.StreamMetadata{BroadcastID: "broadcast"}, nil +} + +func (f *fakeStreamerService) FetchSpadeURL(context.Context, string) (string, error) { + return "https://spade.test", nil +} + +func (f *fakeStreamerService) TouchPlayback(context.Context, string, *twitch.PlaybackToken) error { + return nil +} + +func (f *fakeStreamerService) SendMinuteWatched(context.Context, string, twitch.MinuteWatchedPayload) error { + return nil +} + type fakeIRCSyncer struct { calls [][]string } @@ -71,6 +95,21 @@ func (f *fakeIRCSyncer) Sync(_ context.Context, _, _ string, targets []irc.Targe f.calls = append(f.calls, logins) } +type fakeMinerSyncer struct { + calls [][]string +} + +func (f *fakeMinerSyncer) Sync(_ context.Context, _ *auth.State, _ *twitch.Viewer, entries []twitch.StreamerEntry) { + logins := make([]string, 0, len(entries)) + for _, entry := range entries { + if entry.ChannelID == "" { + continue + } + logins = append(logins, entry.ChannelID) + } + f.calls = append(f.calls, logins) +} + func TestResolveStreamerEntries(t *testing.T) { service := streamService( channel("alpha", "1", "alpha_live"), @@ -78,7 +117,7 @@ func TestResolveStreamerEntries(t *testing.T) { live("1", true), ) - updates := collectUpdates(t, service, []string{"alpha", "beta"}, nil, cancelAfter(0)) + updates := collectUpdates(t, service, []string{"alpha", "beta"}, nil, nil, cancelAfter(0)) if len(updates) != 3 { t.Fatalf("len(updates) = %d", len(updates)) } @@ -98,7 +137,7 @@ func TestResolveStreamerEntriesPlaybackFailureStillMarksLive(t *testing.T) { channel("alpha", "1", "alpha_live"), live("1", true), playbackErr("alpha_live", errors.New("token lookup failed")), - ), []string{"alpha"}, nil, cancelAfter(0)) + ), []string{"alpha"}, nil, nil, cancelAfter(0)) if entry := updates[1].Entry; entry == nil || entry.Status != twitch.StreamerReady || !entry.Live { t.Fatalf("entry = %#v", entry) @@ -109,7 +148,7 @@ func TestResolveStreamerEntriesRefreshesLiveState(t *testing.T) { updates := collectUpdates(t, streamService( channel("alpha", "1", "alpha_live"), streams("1", false, true), - ), []string{"alpha"}, nil, cancelAfter(1)) + ), []string{"alpha"}, nil, nil, cancelAfter(1)) if len(updates) != 3 { t.Fatalf("len(updates) = %d", len(updates)) @@ -124,6 +163,7 @@ func TestResolveStreamerEntriesRefreshesLiveState(t *testing.T) { func TestResolveStreamerEntriesSyncsTopTwoLiveChannels(t *testing.T) { syncer := &fakeIRCSyncer{} + miner := &fakeMinerSyncer{} collectUpdates(t, streamService( channel("alpha", "1", "alpha_live"), channel("beta", "2", "beta_live"), @@ -131,7 +171,7 @@ func TestResolveStreamerEntriesSyncsTopTwoLiveChannels(t *testing.T) { live("1", true), live("2", true), live("3", true), - ), []string{"alpha", "beta", "gamma"}, syncer, cancelAfter(0)) + ), []string{"alpha", "beta", "gamma"}, syncer, miner, cancelAfter(0)) assertSyncCalls(t, syncer.calls, [][]string{ {}, @@ -139,16 +179,23 @@ func TestResolveStreamerEntriesSyncsTopTwoLiveChannels(t *testing.T) { {"alpha_live", "beta_live"}, {"alpha_live", "beta_live"}, }) + assertSyncCalls(t, miner.calls, [][]string{ + {}, + {"1"}, + {"1", "2"}, + {"1", "2", "3"}, + }) } func TestResolveStreamerEntriesRefreshUpdatesWatchedChannels(t *testing.T) { syncer := &fakeIRCSyncer{} + miner := &fakeMinerSyncer{} collectUpdates(t, streamService( channel("alpha", "1", "alpha_live"), channel("beta", "2", "beta_live"), streams("1", true, false), streams("2", false, true), - ), []string{"alpha", "beta"}, syncer, cancelAfter(1)) + ), []string{"alpha", "beta"}, syncer, miner, cancelAfter(1)) assertSyncCalls(t, syncer.calls, [][]string{ {}, @@ -205,10 +252,10 @@ func playbackErr(login string, err error) serviceOption { } } -func collectUpdates(t *testing.T, service *fakeStreamerService, logins []string, syncer ircSyncer, sleep func(context.Context, time.Duration) error) []StreamerUpdate { +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 - err := resolveStreamerEntriesWithSleep(context.Background(), service, &auth.State{Login: "viewer", AccessToken: "token"}, logins, syncer, func(update StreamerUpdate) { + err := resolveStreamerEntriesWithSleep(context.Background(), service, &auth.State{Login: "viewer", AccessToken: "token"}, logins, ircSyncer, minerSyncer, func(update StreamerUpdate) { updates = append(updates, update) }, 0, sleep) if !errors.Is(err, context.Canceled) { diff --git a/internal/twitch/service.go b/internal/twitch/service.go index 7cd7765..acce753 100644 --- a/internal/twitch/service.go +++ b/internal/twitch/service.go @@ -4,13 +4,32 @@ package twitch import ( + "bytes" + "encoding/base64" + "encoding/json" "context" "errors" "fmt" + "io" + "net/http" + "regexp" + "strings" "parasocial/internal/gql" ) +const ( + browserUserAgent = "Mozilla/5.0 (X11; Linux x86_64; rv:85.0) Gecko/20100101 Firefox/85.0" + spadeSettingsRegex = `(https://static\.twitchcdn\.net/config/settings.*?js|https://assets\.twitch\.tv/config/settings.*?\.js)` + spadeURLRegex = `"spade_url":"(.*?)"` +) + +var ( + defaultPageURL = "https://www.twitch.tv/" + settingsAssetPattern = regexp.MustCompile(spadeSettingsRegex) + spadeURLPattern = regexp.MustCompile(spadeURLRegex) +) + // GQLClient is the minimal GraphQL interface the Twitch service depends on. type GQLClient interface { Do(context.Context, gql.Request, any) error @@ -18,7 +37,9 @@ type GQLClient interface { // Service exposes the Twitch lookups the current app needs. type Service struct { - GQL GQLClient + GQL GQLClient + HTTPClient *http.Client + Session gql.Session } // Viewer is the authenticated Twitch account. @@ -44,6 +65,54 @@ type PlaybackToken struct { Value string } +// ChannelPointsContext describes current balance and any immediately claimable bonus chest. +type ChannelPointsContext struct { + Balance int + ClaimID string +} + +// Game describes the current Twitch category for a live stream. +type Game struct { + ID string + Name string + DisplayName string +} + +// StreamMetadata describes the live stream state used for watch telemetry. +type StreamMetadata struct { + BroadcastID string + Title string + Game *Game + ViewerCount int + Tags []Tag + ChannelLogin string +} + +// Tag is one stream tag entry returned by Twitch. +type Tag struct { + ID string + LocalizedName string +} + +// MinuteWatchedPayload is the Spade telemetry body Twitch accepts for simulated viewing. +type MinuteWatchedPayload []minuteWatchedEvent + +type minuteWatchedEvent struct { + Event string `json:"event"` + Properties minuteWatchedProperties `json:"properties"` +} + +type minuteWatchedProperties struct { + ChannelID string `json:"channel_id"` + BroadcastID string `json:"broadcast_id"` + Player string `json:"player"` + UserID string `json:"user_id"` + Live bool `json:"live"` + Channel string `json:"channel"` + Game string `json:"game,omitempty"` + GameID string `json:"game_id,omitempty"` +} + // StreamerStatus describes one row's resolution state in the UI. type StreamerStatus string @@ -159,3 +228,288 @@ func (s *Service) PlaybackAccessToken(ctx context.Context, login string) (*Playb Value: data.StreamPlaybackAccessToken.Value, }, nil } + +// LoadChannelPointsContext fetches one streamer's current balance and any available bonus claim. +func (s *Service) LoadChannelPointsContext(ctx context.Context, login string) (*ChannelPointsContext, error) { + var data struct { + Community *struct { + Channel *struct { + Self *struct { + CommunityPoints *struct { + Balance int `json:"balance"` + AvailableClaim *struct { + ID string `json:"id"` + } `json:"availableClaim"` + } `json:"communityPoints"` + } `json:"self"` + } `json:"channel"` + } `json:"community"` + } + if err := s.GQL.Do(ctx, gql.ChannelPointsContext(login), &data); err != nil { + return nil, err + } + if data.Community == nil || data.Community.Channel == nil || data.Community.Channel.Self == nil || data.Community.Channel.Self.CommunityPoints == nil { + return nil, fmt.Errorf("channel points context missing community points for %s", login) + } + context := &ChannelPointsContext{Balance: data.Community.Channel.Self.CommunityPoints.Balance} + if claim := data.Community.Channel.Self.CommunityPoints.AvailableClaim; claim != nil { + context.ClaimID = claim.ID + } + return context, nil +} + +// ClaimCommunityPoints claims one available channel points bonus chest. +func (s *Service) ClaimCommunityPoints(ctx context.Context, channelID, claimID string) error { + var data struct { + ClaimCommunityPoints *struct { + ID string `json:"id"` + } `json:"claimCommunityPoints"` + } + if err := s.GQL.Do(ctx, gql.ClaimCommunityPoints(channelID, claimID), &data); err != nil { + return err + } + if data.ClaimCommunityPoints == nil { + return fmt.Errorf("claim community points missing response for channel %s", channelID) + } + return nil +} + +// StreamMetadata fetches live stream metadata used to build minute-watched telemetry. +func (s *Service) StreamMetadata(ctx context.Context, login string) (*StreamMetadata, error) { + var data struct { + User *struct { + BroadcastSettings *struct { + Title string `json:"title"` + Game *struct { + ID string `json:"id"` + Name string `json:"name"` + DisplayName string `json:"displayName"` + } `json:"game"` + } `json:"broadcastSettings"` + Stream *struct { + ID string `json:"id"` + ViewersCount int `json:"viewersCount"` + Tags []struct { + ID string `json:"id"` + LocalizedName string `json:"localizedName"` + } `json:"tags"` + } `json:"stream"` + } `json:"user"` + } + if err := s.GQL.Do(ctx, gql.VideoPlayerStreamInfoOverlayChannel(login), &data); err != nil { + return nil, err + } + if data.User == nil || data.User.Stream == nil { + return nil, fmt.Errorf("stream metadata missing live stream for %s", login) + } + if data.User.BroadcastSettings == nil { + return nil, fmt.Errorf("stream metadata missing broadcast settings for %s", login) + } + + metadata := &StreamMetadata{ + BroadcastID: data.User.Stream.ID, + Title: strings.TrimSpace(data.User.BroadcastSettings.Title), + ViewerCount: data.User.Stream.ViewersCount, + ChannelLogin: login, + } + if game := data.User.BroadcastSettings.Game; game != nil { + metadata.Game = &Game{ + ID: game.ID, + Name: game.Name, + DisplayName: game.DisplayName, + } + } + for _, tag := range data.User.Stream.Tags { + metadata.Tags = append(metadata.Tags, Tag{ + ID: tag.ID, + LocalizedName: tag.LocalizedName, + }) + } + if metadata.BroadcastID == "" { + return nil, fmt.Errorf("stream metadata missing broadcast id for %s", login) + } + return metadata, nil +} + +// FetchSpadeURL loads the public Twitch page and extracts the Spade telemetry endpoint. +func (s *Service) FetchSpadeURL(ctx context.Context, login string) (string, error) { + pageReq, err := http.NewRequestWithContext(ctx, http.MethodGet, defaultPageURL+login, nil) + if err != nil { + return "", err + } + pageReq.Header.Set("User-Agent", browserUserAgent) + pageResp, err := s.httpClient().Do(pageReq) + if err != nil { + return "", fmt.Errorf("fetch streamer page: %w", err) + } + defer pageResp.Body.Close() + pageBody, err := io.ReadAll(pageResp.Body) + if err != nil { + return "", fmt.Errorf("read streamer page: %w", err) + } + if pageResp.StatusCode != http.StatusOK { + return "", fmt.Errorf("fetch streamer page: status %d", pageResp.StatusCode) + } + settingsURL := settingsAssetPattern.FindString(string(pageBody)) + if settingsURL == "" { + return "", fmt.Errorf("streamer page missing settings asset for %s", login) + } + + settingsReq, err := http.NewRequestWithContext(ctx, http.MethodGet, settingsURL, nil) + if err != nil { + return "", err + } + settingsReq.Header.Set("User-Agent", browserUserAgent) + settingsResp, err := s.httpClient().Do(settingsReq) + if err != nil { + return "", fmt.Errorf("fetch settings asset: %w", err) + } + defer settingsResp.Body.Close() + settingsBody, err := io.ReadAll(settingsResp.Body) + if err != nil { + return "", fmt.Errorf("read settings asset: %w", err) + } + if settingsResp.StatusCode != http.StatusOK { + return "", fmt.Errorf("fetch settings asset: status %d", settingsResp.StatusCode) + } + + match := spadeURLPattern.FindSubmatch(settingsBody) + if len(match) < 2 { + return "", fmt.Errorf("settings asset missing spade_url for %s", login) + } + return strings.ReplaceAll(string(match[1]), `\/`, `/`), nil +} + +// TouchPlayback walks the HLS playlist chain Twitch expects before telemetry is posted. +func (s *Service) TouchPlayback(ctx context.Context, login string, token *PlaybackToken) error { + if token == nil || token.Signature == "" || token.Value == "" { + return errors.New("playback access token is incomplete") + } + masterURL := gql.HLSMasterPlaylistURL(login, token.Signature, token.Value) + variantURL, err := s.fetchPlaylistTarget(ctx, masterURL) + if err != nil { + return fmt.Errorf("fetch master playlist: %w", err) + } + mediaURL, err := s.fetchPlaylistTarget(ctx, variantURL) + if err != nil { + return fmt.Errorf("fetch variant playlist: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodHead, mediaURL, nil) + if err != nil { + return err + } + req.Header.Set("User-Agent", s.requestUserAgent()) + resp, err := s.httpClient().Do(req) + if err != nil { + return fmt.Errorf("head media playlist: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("head media playlist: status %d", resp.StatusCode) + } + return nil +} + +// BuildMinuteWatchedPayload constructs the Spade telemetry payload for one live stream. +func BuildMinuteWatchedPayload(userID, channelID, login string, metadata *StreamMetadata) MinuteWatchedPayload { + broadcastID := "" + if metadata != nil { + broadcastID = metadata.BroadcastID + } + payload := MinuteWatchedPayload{ + { + Event: "minute-watched", + Properties: minuteWatchedProperties{ + ChannelID: channelID, + BroadcastID: broadcastID, + Player: "site", + UserID: userID, + Live: true, + Channel: login, + }, + }, + } + if metadata != nil && metadata.Game != nil && metadata.Game.Name != "" && metadata.Game.ID != "" { + payload[0].Properties.Game = metadata.Game.Name + payload[0].Properties.GameID = metadata.Game.ID + } + return payload +} + +// Encode returns the form body Twitch expects for Spade telemetry. +func (p MinuteWatchedPayload) Encode() (string, error) { + body, err := json.Marshal(p) + if err != nil { + return "", err + } + return base64.StdEncoding.EncodeToString(body), nil +} + +// SendMinuteWatched posts one minute-watched payload to Twitch's Spade endpoint. +func (s *Service) SendMinuteWatched(ctx context.Context, spadeURL string, payload MinuteWatchedPayload) error { + encoded, err := payload.Encode() + if err != nil { + return fmt.Errorf("encode minute watched payload: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, spadeURL, strings.NewReader("data="+encoded)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("User-Agent", s.requestUserAgent()) + resp, err := s.httpClient().Do(req) + if err != nil { + return fmt.Errorf("post minute watched payload: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("post minute watched payload: status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + return nil +} + +func (s *Service) fetchPlaylistTarget(ctx context.Context, targetURL string) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, targetURL, nil) + if err != nil { + return "", err + } + req.Header.Set("User-Agent", s.requestUserAgent()) + resp, err := s.httpClient().Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("status %d", resp.StatusCode) + } + + lines := bytes.Split(body, []byte{'\n'}) + for i := len(lines) - 1; i >= 0; i-- { + line := strings.TrimSpace(string(lines[i])) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + return line, nil + } + return "", errors.New("playlist contained no media target") +} + +func (s *Service) httpClient() *http.Client { + if s.HTTPClient != nil { + return s.HTTPClient + } + return http.DefaultClient +} + +func (s *Service) requestUserAgent() string { + if s.Session.UserAgent != "" { + return s.Session.UserAgent + } + return browserUserAgent +} diff --git a/internal/twitch/service_test.go b/internal/twitch/service_test.go index 34873e5..254049b 100644 --- a/internal/twitch/service_test.go +++ b/internal/twitch/service_test.go @@ -2,8 +2,13 @@ package twitch import ( "context" + "encoding/base64" "encoding/json" "errors" + "io" + "net/http" + "net/http/httptest" + "strings" "testing" "parasocial/internal/gql" @@ -124,3 +129,176 @@ func TestPlaybackAccessTokenMissingFields(t *testing.T) { t.Fatal("expected error") } } + +func TestLoadChannelPointsContext(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "ChannelPointsContext": `{"community":{"channel":{"self":{"communityPoints":{"balance":1234,"availableClaim":{"id":"claim-1"}}}}}}`, + }} + service := &Service{GQL: client} + context, err := service.LoadChannelPointsContext(context.Background(), "streamer") + if err != nil { + t.Fatal(err) + } + if context.Balance != 1234 || context.ClaimID != "claim-1" { + t.Fatalf("context = %#v", context) + } +} + +func TestClaimCommunityPoints(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "ClaimCommunityPoints": `{"claimCommunityPoints":{"id":"claim-1"}}`, + }} + service := &Service{GQL: client} + if err := service.ClaimCommunityPoints(context.Background(), "7", "claim-1"); err != nil { + t.Fatal(err) + } + if len(client.requests) != 1 { + t.Fatalf("requests = %d", len(client.requests)) + } + input := client.requests[0].Variables["input"].(map[string]any) + if input["channelID"] != "7" || input["claimID"] != "claim-1" { + t.Fatalf("input = %#v", input) + } +} + +func TestStreamMetadata(t *testing.T) { + t.Parallel() + + client := &fakeGQL{data: map[string]string{ + "VideoPlayerStreamInfoOverlayChannel": `{"user":{"broadcastSettings":{"title":" Live title ","game":{"id":"99","name":"game-name","displayName":"Game Name"}},"stream":{"id":"broadcast","viewersCount":77,"tags":[{"id":"1","localizedName":"English"}]}}}`, + }} + service := &Service{GQL: client} + metadata, err := service.StreamMetadata(context.Background(), "streamer") + if err != nil { + t.Fatal(err) + } + if metadata.BroadcastID != "broadcast" || metadata.Title != "Live title" { + t.Fatalf("metadata = %#v", metadata) + } + if metadata.Game == nil || metadata.Game.Name != "game-name" { + t.Fatalf("game = %#v", metadata.Game) + } + if len(metadata.Tags) != 1 || metadata.Tags[0].LocalizedName != "English" { + t.Fatalf("tags = %#v", metadata.Tags) + } +} + +func TestFetchSpadeURL(t *testing.T) { + t.Parallel() + + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/streamer": + _, _ = io.WriteString(w, ``) + case "/config/settings.abc.js": + _, _ = io.WriteString(w, `{"spade_url":"https:\/\/spade.example.test\/collect"}`) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + originalPageURL := defaultPageURL + defaultPageURL = server.URL + "/" + defer func() { defaultPageURL = originalPageURL }() + + service := &Service{HTTPClient: rewriteClient(server.URL, server.Client())} + spadeURL, err := service.FetchSpadeURL(context.Background(), "streamer") + if err != nil { + t.Fatal(err) + } + if spadeURL != "https://spade.example.test/collect" { + t.Fatalf("spadeURL = %q", spadeURL) + } +} + +func TestTouchPlayback(t *testing.T) { + t.Parallel() + + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/channel/hls/streamer.m3u8": + _, _ = io.WriteString(w, "#EXTM3U\n"+server.URL+"/variant.m3u8\n") + case "/variant.m3u8": + _, _ = io.WriteString(w, "#EXTM3U\n#EXTINF:1,\n"+server.URL+"/chunk.ts\n") + case "/chunk.ts": + if r.Method != http.MethodHead { + t.Fatalf("method = %s, want HEAD", r.Method) + } + w.WriteHeader(http.StatusOK) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + service := &Service{HTTPClient: rewriteClient(server.URL, server.Client())} + if err := service.TouchPlayback(context.Background(), "streamer", &PlaybackToken{Signature: "sig", Value: "token"}); err != nil { + t.Fatal(err) + } +} + +func TestSendMinuteWatched(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s", r.Method) + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatal(err) + } + encoded := strings.TrimPrefix(string(body), "data=") + decoded, err := base64.StdEncoding.DecodeString(encoded) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(decoded), `"event":"minute-watched"`) { + t.Fatalf("payload = %s", decoded) + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + service := &Service{HTTPClient: server.Client()} + payload := BuildMinuteWatchedPayload("viewer", "channel", "streamer", &StreamMetadata{ + BroadcastID: "broadcast", + Game: &Game{ID: "game-id", Name: "game-name"}, + }) + if err := service.SendMinuteWatched(context.Background(), server.URL, payload); err != nil { + t.Fatal(err) + } +} + +func rewriteClient(baseURL string, client *http.Client) *http.Client { + copy := *client + copy.Transport = rewriteTransport{baseURL: baseURL, next: client.Transport} + return © +} + +type rewriteTransport struct { + baseURL string + next http.RoundTripper +} + +func (t rewriteTransport) RoundTrip(r *http.Request) (*http.Response, error) { + if strings.HasPrefix(r.URL.Host, "usher.ttvnw.net") || strings.HasPrefix(r.URL.Host, "assets.twitch.tv") || strings.HasPrefix(r.URL.Host, "static.twitchcdn.net") { + target, err := http.NewRequest(http.MethodGet, t.baseURL, nil) + if err != nil { + return nil, err + } + r.URL.Scheme = target.URL.Scheme + r.URL.Host = target.URL.Host + } + if next := t.next; next != nil { + return next.RoundTrip(r) + } + return http.DefaultTransport.RoundTrip(r) +}