feat: add twitch channel points miner runtime
This commit is contained in:
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user