diff --git a/cmd/parasocial/main.go b/cmd/parasocial/main.go index 2557ec3..eda1d3a 100644 --- a/cmd/parasocial/main.go +++ b/cmd/parasocial/main.go @@ -1,6 +1,5 @@ // main.go is the executable entrypoint for the rewritten CLI. -// It owns process-level concerns such as signal handling, exit codes, -// and delegating the actual application startup to internal/app. +// It owns process-level concerns such as signal handling and exit codes. package main import ( @@ -11,7 +10,8 @@ import ( "os/signal" "syscall" - "parasocial/internal/app" + "parasocial/internal/config" + "parasocial/internal/tui" ) // main sets up process cancellation and reports fatal startup errors to stderr. @@ -19,7 +19,11 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - if err := app.Run(ctx); err != nil { + cfg, err := config.Load("config.toml") + if err == nil { + err = tui.Run(ctx, tui.Options{Streamers: cfg.Streamers}) + } + if err != nil { if errors.Is(err, context.Canceled) { return } diff --git a/go.mod b/go.mod index 2f2c5f6..03aae4d 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/charmbracelet/bubbles v1.0.0 github.com/charmbracelet/bubbletea v1.3.10 github.com/charmbracelet/lipgloss v1.1.0 + github.com/gorilla/websocket v1.5.3 ) require ( @@ -19,7 +20,6 @@ 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/internal/app/app.go b/internal/app/app.go deleted file mode 100644 index 53abbb1..0000000 --- a/internal/app/app.go +++ /dev/null @@ -1,18 +0,0 @@ -// app.go owns process-level application bootstrap before handing control to the TUI. -package app - -import ( - "context" - - "parasocial/internal/config" - "parasocial/internal/tui" -) - -// Run loads configuration and starts the terminal UI. -func Run(ctx context.Context) error { - cfg, err := config.LoadDefault() - if err != nil { - return err - } - return tui.Run(ctx, tui.Options{Streamers: cfg.Streamers}) -} diff --git a/internal/auth/auth.go b/internal/auth/auth.go index 176bb51..14de7f1 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -13,6 +13,7 @@ import ( "math/big" "net/http" "net/url" + "slices" "strings" "time" ) @@ -27,7 +28,7 @@ const ( tvOrigin = "https://android.tv.twitch.tv" tvReferer = "https://android.tv.twitch.tv/" - tvUserAgent = "Mozilla/5.0 (Linux; Android 7.1; Smart Box C1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36" + TVUserAgent = "Mozilla/5.0 (Linux; Android 7.1; Smart Box C1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36" ) var RequiredScopes = []string{ @@ -71,20 +72,15 @@ type deviceCodeResponse struct { // tokenResponse models the token payload returned after device authorization succeeds. type tokenResponse struct { - AccessToken string `json:"access_token"` - RefreshToken string `json:"refresh_token"` - Scope []string `json:"scope"` - TokenType string `json:"token_type"` - ExpiresIn int `json:"expires_in"` + AccessToken string `json:"access_token"` } // validationResponse models the token introspection data returned by Twitch validate. type validationResponse struct { - ClientID string `json:"client_id"` - Login string `json:"login"` - Scopes []string `json:"scopes"` - UserID string `json:"user_id"` - ExpiresIn int `json:"expires_in"` + ClientID string `json:"client_id"` + Login string `json:"login"` + Scopes []string `json:"scopes"` + UserID string `json:"user_id"` } // oauthErrorResponse captures structured OAuth errors returned by Twitch endpoints. @@ -147,7 +143,7 @@ func (c *Client) ReuseAuth(ctx context.Context, path string) (*State, error) { return nil, nil } - state.applyValidation(validation, c.now()) + state.Login = validation.Login if err := SaveState(path, state); err != nil { return nil, fmt.Errorf("save validated auth state: %w", err) } @@ -175,7 +171,7 @@ func (c *Client) EnsureAuth(ctx context.Context, path string, status StatusFunc) c.status(status, "Validating cached token from %s", path) validation, err := c.ValidateToken(ctx, state.AccessToken) if err == nil && hasAllScopes(validation.Scopes, RequiredScopes) { - state.applyValidation(validation, c.now()) + state.Login = validation.Login if err := SaveState(path, state); err != nil { return nil, fmt.Errorf("save validated auth state: %w", err) } @@ -203,14 +199,7 @@ func (c *Client) EnsureAuth(ctx context.Context, path string, status StatusFunc) } state.AccessToken = tokenResp.AccessToken - state.RefreshToken = tokenResp.RefreshToken - state.TokenType = tokenResp.TokenType - if state.TokenType == "" { - state.TokenType = "bearer" - } - state.Scopes = tokenResp.Scope - state.ExpiresIn = tokenResp.ExpiresIn - state.applyValidation(validation, c.now()) + state.Login = validation.Login if err := SaveState(path, state); err != nil { return nil, fmt.Errorf("save auth state: %w", err) @@ -222,13 +211,13 @@ func (c *Client) EnsureAuth(ctx context.Context, path string, status StatusFunc) // ValidateToken asks Twitch whether an OAuth access token is still valid. func (c *Client) ValidateToken(ctx context.Context, accessToken string) (*validationResponse, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.endpoints().Validate, nil) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.Endpoints.Validate, nil) if err != nil { return nil, err } req.Header.Set("Authorization", "OAuth "+accessToken) - resp, err := c.httpClient().Do(req) + resp, err := c.HTTPClient.Do(req) if err != nil { return nil, err } @@ -269,7 +258,7 @@ func (c *Client) activate(ctx context.Context, deviceID string, status StatusFun if interval <= 0 { interval = 5 * time.Second } - deadline := c.now().Add(time.Duration(deviceResp.ExpiresIn) * time.Second) + deadline := c.Now().Add(time.Duration(deviceResp.ExpiresIn) * time.Second) c.status(status, "== Twitch activation ==") c.status(status, "Open page: %s", deviceResp.VerificationURI) @@ -286,7 +275,7 @@ func (c *Client) requestDeviceCode(ctx context.Context, deviceID string) (*devic form.Set("client_id", ClientID) form.Set("scopes", strings.Join(RequiredScopes, " ")) - body, statusCode, err := c.postForm(ctx, c.endpoints().Device, deviceID, form) + body, statusCode, err := c.postForm(ctx, c.Endpoints.Device, deviceID, form) if err != nil { return nil, err } @@ -311,13 +300,13 @@ func (c *Client) requestDeviceCode(ctx context.Context, deviceID string) (*devic func (c *Client) pollForToken(ctx context.Context, deviceID, deviceCode string, interval time.Duration, deadline time.Time, status StatusFunc) (*tokenResponse, error) { attempt := 1 for { - if !c.now().Before(deadline) { + if !c.Now().Before(deadline) { return nil, fmt.Errorf("device code expired at %s before authorization completed", deadline.Format(time.RFC3339)) } - remaining := deadline.Sub(c.now()).Round(time.Second) + remaining := deadline.Sub(c.Now()).Round(time.Second) c.status(status, "[poll %d] waiting %s before token request (%s remaining)", attempt, interval, remaining) - if err := c.sleep(ctx, interval); err != nil { + if err := c.Sleep(ctx, interval); err != nil { return nil, err } @@ -326,7 +315,7 @@ func (c *Client) pollForToken(ctx context.Context, deviceID, deviceCode string, form.Set("device_code", deviceCode) form.Set("grant_type", pollGrantType) - body, statusCode, err := c.postForm(ctx, c.endpoints().Token, deviceID, form) + body, statusCode, err := c.postForm(ctx, c.Endpoints.Token, deviceID, form) if err != nil { return nil, fmt.Errorf("poll request failed on attempt %d: %w", attempt, err) } @@ -368,12 +357,18 @@ func (c *Client) postForm(ctx context.Context, endpoint, deviceID string, form u if err != nil { return nil, 0, err } - for key, value := range defaultOAuthHeaders(deviceID) { - req.Header.Set(key, value) - } + req.Header.Set("Accept", "application/json") + req.Header.Set("Accept-Language", "en-US") + req.Header.Set("Cache-Control", "no-cache") + req.Header.Set("Client-Id", ClientID) + req.Header.Set("Origin", tvOrigin) + req.Header.Set("Pragma", "no-cache") + req.Header.Set("Referer", tvReferer) + req.Header.Set("User-Agent", TVUserAgent) + req.Header.Set("X-Device-Id", deviceID) req.Header.Set("Content-Type", contentTypeForm) - resp, err := c.httpClient().Do(req) + resp, err := c.HTTPClient.Do(req) if err != nil { return nil, 0, err } @@ -431,51 +426,10 @@ func (r oauthErrorResponse) summary() string { return strings.Join(parts, ", ") } -// defaultOAuthHeaders builds the Twitch TV request headers used for auth requests. -func defaultOAuthHeaders(deviceID string) map[string]string { - return map[string]string{ - "Accept": "application/json", - "Accept-Language": "en-US", - "Cache-Control": "no-cache", - "Client-Id": ClientID, - "Origin": tvOrigin, - "Pragma": "no-cache", - "Referer": tvReferer, - "User-Agent": TVUserAgent(), - "X-Device-Id": deviceID, - } -} - -// TVUserAgent returns the Twitch TV user agent used to impersonate the device client. -func TVUserAgent() string { - return tvUserAgent -} - -// applyValidation copies validated account metadata back onto the persisted auth state. -func (s *State) applyValidation(validation *validationResponse, validatedAt time.Time) { - s.Login = validation.Login - s.UserID = validation.UserID - s.ClientID = validation.ClientID - s.Scopes = validation.Scopes - s.ExpiresIn = validation.ExpiresIn - s.ValidatedAt = validatedAt - s.Cookies = s.persistedCookies() -} - -// hasScope reports whether the token scopes include one exact scope string. -func hasScope(scopes []string, scope string) bool { - for _, candidate := range scopes { - if candidate == scope { - return true - } - } - return false -} - // hasAllScopes reports whether the token satisfies the full required scope set. func hasAllScopes(scopes []string, required []string) bool { for _, scope := range required { - if !hasScope(scopes, scope) { + if !slices.Contains(scopes, scope) { return false } } @@ -511,38 +465,6 @@ func sleepContext(ctx context.Context, d time.Duration) error { } } -// httpClient returns the configured HTTP client or falls back to the default client. -func (c *Client) httpClient() *http.Client { - if c.HTTPClient != nil { - return c.HTTPClient - } - return http.DefaultClient -} - -// endpoints returns the configured endpoint set or the package defaults. -func (c *Client) endpoints() Endpoints { - if c.Endpoints == (Endpoints{}) { - return DefaultEndpoints - } - return c.Endpoints -} - -// now returns the current time using the injected clock when one is configured. -func (c *Client) now() time.Time { - if c.Now != nil { - return c.Now() - } - return time.Now() -} - -// sleep delegates waiting to the injected hook when tests need deterministic timing. -func (c *Client) sleep(ctx context.Context, d time.Duration) error { - if c.Sleep != nil { - return c.Sleep(ctx, d) - } - return sleepContext(ctx, d) -} - // status formats and emits one auth progress line when a status sink is attached. func (c *Client) status(status StatusFunc, format string, args ...any) { if status == nil { diff --git a/internal/auth/auth_test.go b/internal/auth/auth_test.go index 1f1acc1..99babb1 100644 --- a/internal/auth/auth_test.go +++ b/internal/auth/auth_test.go @@ -9,24 +9,18 @@ import ( "net/url" "os" "path/filepath" - "slices" "strings" "testing" "time" ) -func TestSaveStatePersistsCookieEntries(t *testing.T) { +func TestSaveAndLoadState(t *testing.T) { t.Parallel() path := filepath.Join(t.TempDir(), "cookies.json") state := &State{ AccessToken: "token", - TokenType: "bearer", - Scopes: RequiredScopes, Login: "viewer", - UserID: "123", - ClientID: ClientID, - ExpiresIn: 3600, DeviceID: "device", } if err := SaveState(path, state); err != nil { @@ -37,44 +31,8 @@ func TestSaveStatePersistsCookieEntries(t *testing.T) { if err != nil { t.Fatal(err) } - if loaded.AccessToken != "token" { - t.Fatalf("AccessToken = %q", loaded.AccessToken) - } - - gotNames := []string{} - for _, cookie := range loaded.Cookies { - gotNames = append(gotNames, cookie.Name) - } - wantNames := []string{"auth-token", "login", "persistent"} - if !slices.Equal(gotNames, wantNames) { - t.Fatalf("cookie names = %#v, want %#v", gotNames, wantNames) - } -} - -func TestSaveStateAllowsZeroExpiresIn(t *testing.T) { - t.Parallel() - - path := filepath.Join(t.TempDir(), "cookies.json") - state := &State{ - AccessToken: "token", - TokenType: "bearer", - Scopes: RequiredScopes, - Login: "viewer", - UserID: "123", - ClientID: ClientID, - ExpiresIn: 0, - DeviceID: "device", - } - if err := SaveState(path, state); err != nil { - t.Fatalf("SaveState() error = %v", err) - } - - loaded, err := LoadState(path) - if err != nil { - t.Fatal(err) - } - if loaded.ExpiresIn != 0 { - t.Fatalf("ExpiresIn = %d, want 0", loaded.ExpiresIn) + if *loaded != *state { + t.Fatalf("state = %#v, want %#v", loaded, state) } } @@ -102,7 +60,7 @@ func TestLoadStateRejectsPartialAuthBundle(t *testing.T) { if err == nil { t.Fatal("LoadState() error = nil, want error") } - if !strings.Contains(err.Error(), "missing token_type") { + if !strings.Contains(err.Error(), "missing login") { t.Fatalf("error = %q", err) } } @@ -120,11 +78,10 @@ func TestReuseAuthReusesValidCachedToken(t *testing.T) { t.Fatalf("Authorization = %q", got) } writeJSON(t, w, validationResponse{ - ClientID: ClientID, - Login: "viewer", - UserID: "123", - Scopes: RequiredScopes, - ExpiresIn: 3600, + ClientID: ClientID, + Login: "viewer", + UserID: "123", + Scopes: RequiredScopes, }) })) defer server.Close() @@ -132,12 +89,7 @@ func TestReuseAuthReusesValidCachedToken(t *testing.T) { tokenFile := filepath.Join(t.TempDir(), "cookies.json") if err := SaveState(tokenFile, &State{ AccessToken: "cached-token", - TokenType: "bearer", - Scopes: RequiredScopes, Login: "viewer", - UserID: "123", - ClientID: ClientID, - ExpiresIn: 3600, DeviceID: "device", }); err != nil { t.Fatal(err) @@ -145,8 +97,6 @@ func TestReuseAuthReusesValidCachedToken(t *testing.T) { client := NewClient(server.Client()) client.Endpoints = Endpoints{Validate: server.URL + "/validate"} - client.Now = func() time.Time { return time.Unix(100, 0).UTC() } - state, err := client.ReuseAuth(context.Background(), tokenFile) if err != nil { t.Fatal(err) @@ -169,11 +119,10 @@ func TestEnsureAuthActivatesWhenCachedTokenMissingChatRead(t *testing.T) { validateCalls++ if validateCalls == 1 { writeJSON(t, w, validationResponse{ - ClientID: ClientID, - Login: "viewer", - UserID: "123", - Scopes: []string{"user:read:email"}, - ExpiresIn: 100, + ClientID: ClientID, + Login: "viewer", + UserID: "123", + Scopes: []string{"user:read:email"}, }) return } @@ -181,11 +130,10 @@ func TestEnsureAuthActivatesWhenCachedTokenMissingChatRead(t *testing.T) { t.Fatalf("Authorization = %q", got) } writeJSON(t, w, validationResponse{ - ClientID: ClientID, - Login: "viewer", - UserID: "123", - Scopes: RequiredScopes, - ExpiresIn: 3600, + ClientID: ClientID, + Login: "viewer", + UserID: "123", + Scopes: RequiredScopes, }) case "/device": if err := r.ParseForm(); err != nil { @@ -214,11 +162,7 @@ func TestEnsureAuthActivatesWhenCachedTokenMissingChatRead(t *testing.T) { t.Fatalf("device_code = %q", got) } writeJSON(t, w, tokenResponse{ - AccessToken: "new-token", - RefreshToken: "refresh-token", - Scope: RequiredScopes, - TokenType: "bearer", - ExpiresIn: 3600, + AccessToken: "new-token", }) default: t.Fatalf("unexpected path %s", r.URL.Path) @@ -229,12 +173,7 @@ func TestEnsureAuthActivatesWhenCachedTokenMissingChatRead(t *testing.T) { tokenFile := filepath.Join(t.TempDir(), "cookies.json") if err := SaveState(tokenFile, &State{ AccessToken: "wrong-scope-token", - TokenType: "bearer", - Scopes: []string{"user:read:email"}, Login: "viewer", - UserID: "123", - ClientID: ClientID, - ExpiresIn: 3600, DeviceID: "device", }); err != nil { t.Fatal(err) diff --git a/internal/auth/store.go b/internal/auth/store.go index 3a9c915..82ba967 100644 --- a/internal/auth/store.go +++ b/internal/auth/store.go @@ -9,30 +9,15 @@ import ( "fmt" "os" "path/filepath" - "time" ) const authFileName = "cookies.json" -// Cookie stores one persisted cookie-like name/value entry in the auth bundle. -type Cookie struct { - Name string `json:"name"` - Value string `json:"value"` -} - // State is the persisted cwd auth bundle for the app. type State struct { - AccessToken string `json:"access_token"` - RefreshToken string `json:"refresh_token,omitempty"` - TokenType string `json:"token_type"` - Scopes []string `json:"scopes"` - Login string `json:"login"` - UserID string `json:"user_id"` - ClientID string `json:"client_id"` - ExpiresIn int `json:"expires_in"` - ValidatedAt time.Time `json:"validated_at"` - DeviceID string `json:"device_id"` - Cookies []Cookie `json:"cookies,omitempty"` + AccessToken string `json:"access_token"` + Login string `json:"login"` + DeviceID string `json:"device_id"` } // DefaultPath returns the cwd auth bundle path. @@ -72,10 +57,7 @@ func SaveState(path string, state *State) error { return err } - copy := *state - copy.Cookies = copy.persistedCookies() - - data, err := json.MarshalIndent(copy, "", " ") + data, err := json.MarshalIndent(state, "", " ") if err != nil { return err } @@ -83,30 +65,13 @@ func SaveState(path string, state *State) error { return os.WriteFile(path, data, 0o600) } -// persistedCookies rebuilds the cookie-like entries that must travel with the saved auth bundle. -func (s *State) persistedCookies() []Cookie { - return []Cookie{ - {Name: "auth-token", Value: s.AccessToken}, - {Name: "login", Value: s.Login}, - {Name: "persistent", Value: s.UserID}, - } -} - // validate rejects partial auth bundles so loading and saving use one canonical shape. func (s *State) validate() error { switch { case s.AccessToken == "": return fmt.Errorf("missing access_token") - case s.TokenType == "": - return fmt.Errorf("missing token_type") case s.Login == "": return fmt.Errorf("missing login") - case s.UserID == "": - return fmt.Errorf("missing user_id") - case s.ClientID == "": - return fmt.Errorf("missing client_id") - case len(s.Scopes) == 0: - return fmt.Errorf("missing scopes") case s.DeviceID == "": return fmt.Errorf("missing device_id") default: diff --git a/internal/config/config.go b/internal/config/config.go index cb80bb7..67d16b4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -12,23 +12,11 @@ import ( "github.com/BurntSushi/toml" ) -const configFileName = "config.toml" - // Config is the user-owned TOML configuration for parasocial. type Config struct { Streamers []string `toml:"streamers"` } -// DefaultPath returns the config file path relative to the current working directory. -func DefaultPath() string { - return configFileName -} - -// LoadDefault reads config from the current working directory. -func LoadDefault() (Config, error) { - return Load(DefaultPath()) -} - // Load reads, normalizes, and validates a TOML config file. func Load(path string) (Config, error) { var cfg Config diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 2a953b7..5926ff5 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -46,12 +46,6 @@ func TestLoadRejectsInvalidTOML(t *testing.T) { } } -func TestDefaultPathUsesCurrentWorkingDirectoryConfig(t *testing.T) { - if got := DefaultPath(); got != "config.toml" { - t.Fatalf("DefaultPath() = %q, want %q", got, "config.toml") - } -} - func TestLoadRejectsMissingFile(t *testing.T) { path := filepath.Join(t.TempDir(), "missing.toml") diff --git a/internal/gql/operations.go b/internal/gql/operations.go index 7221242..b741257 100644 --- a/internal/gql/operations.go +++ b/internal/gql/operations.go @@ -3,8 +3,6 @@ // 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"` @@ -120,16 +118,3 @@ func VideoPlayerStreamInfoOverlayChannel(login string) Request { "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/irc/client.go b/internal/irc/client.go index 4f4bc45..8c66a96 100644 --- a/internal/irc/client.go +++ b/internal/irc/client.go @@ -6,7 +6,6 @@ import ( "context" "errors" "fmt" - "io" "net" "strings" "sync" @@ -41,14 +40,12 @@ type Client struct { Token string Streamer string Once bool - Debug bool - Out io.Writer Events EventSink DialContext func(context.Context, string, string) (net.Conn, error) } // Run connects, authenticates, joins the configured streamer channel, and stays connected. -func (c *Client) Run(ctx context.Context) error { +func (c *Client) Run(ctx context.Context) (err error) { if c.Login == "" { return errors.New("missing Twitch login") } @@ -58,23 +55,23 @@ func (c *Client) Run(ctx context.Context) error { if c.Streamer == "" { return errors.New("missing streamer") } + defer func() { + if ctx.Err() != nil { + err = nil + } + }() addr := c.Addr if addr == "" { addr = DefaultAddr } - c.status("Connecting to %s\n", addr) conn, err := c.dial(ctx, addr) if err != nil { return fmt.Errorf("connect to Twitch IRC: %w", err) } - session := &session{ - conn: conn, - out: c.Out, - debug: c.Debug, - } + session := &session{conn: conn} defer session.close() done := make(chan struct{}) @@ -82,24 +79,21 @@ func (c *Client) Run(ctx context.Context) error { go func() { select { case <-ctx.Done(): - _ = session.send("QUIT :parasocial shutting down", false) + _ = session.send("QUIT :parasocial shutting down") session.close() case <-done: } }() - if err := session.send("PASS oauth:"+c.Token, true); err != nil { + if err := session.send("PASS oauth:" + c.Token); err != nil { return err } - c.status("Sent PASS\n") - if err := session.send("NICK "+c.Login, false); err != nil { + if err := session.send("NICK " + c.Login); err != nil { return err } - c.status("Sent NICK %s\n", c.Login) - if err := session.send("JOIN #"+c.Streamer, false); err != nil { + if err := session.send("JOIN #" + c.Streamer); err != nil { return err } - c.status("Sent JOIN #%s\n", c.Streamer) reader := bufio.NewReader(conn) joined := false @@ -116,13 +110,10 @@ func (c *Client) Run(ctx context.Context) error { } line = strings.TrimRight(line, "\r\n") - session.debugIn(line) - if payload, ok := pingPayload(line); ok { - if err := session.send("PONG "+payload, false); err != nil { + if err := session.send("PONG " + payload); err != nil { return err } - c.status("Responded to PING\n") continue } @@ -133,16 +124,14 @@ func (c *Client) Run(ctx context.Context) error { return fmt.Errorf("IRC join denied: %s", line) } if isWelcome(line) { - c.status("IRC authentication accepted\n") continue } if isJoinConfirmation(line, c.Login, c.Streamer) { joinedLine := fmt.Sprintf("Joined #%s as %s", c.Streamer, c.Login) - c.status("%s\n", joinedLine) c.emit(StateJoined, joinedLine) joined = true if c.Once { - if err := session.send("QUIT :parasocial --once complete", false); err != nil { + if err := session.send("QUIT :parasocial --once complete"); err != nil { return err } return nil @@ -150,41 +139,25 @@ func (c *Client) Run(ctx context.Context) error { continue } if joined { - c.status("%s\n", line) c.emit("", line) } } } type session struct { - conn net.Conn - out io.Writer - debug bool - mu sync.Mutex - once sync.Once + conn net.Conn + mu sync.Mutex + once sync.Once } -func (s *session) send(line string, redact bool) error { +func (s *session) send(line string) error { s.mu.Lock() defer s.mu.Unlock() - debugLine := line - if redact { - debugLine = "PASS oauth:" - } - if s.debug && s.out != nil { - fmt.Fprintf(s.out, "> %s\n", debugLine) - } _, err := fmt.Fprintf(s.conn, "%s\r\n", line) return err } -func (s *session) debugIn(line string) { - if s.debug && s.out != nil { - fmt.Fprintf(s.out, "< %s\n", line) - } -} - func (s *session) close() { s.once.Do(func() { _ = s.conn.SetDeadline(time.Now()) @@ -200,13 +173,6 @@ func (c *Client) dial(ctx context.Context, addr string) (net.Conn, error) { return dialer.DialContext(ctx, "tcp", addr) } -func (c *Client) status(format string, args ...any) { - if c.Out == nil { - return - } - fmt.Fprintf(c.Out, format, args...) -} - func (c *Client) emit(state ConnectionState, line string) { if c.Events == nil { return diff --git a/internal/irc/client_test.go b/internal/irc/client_test.go index de542a8..f70ba0c 100644 --- a/internal/irc/client_test.go +++ b/internal/irc/client_test.go @@ -64,7 +64,6 @@ func TestRunOnceSendsAuthJoinAndRespondsToPing(t *testing.T) { Token: "token", Streamer: "streamer", Once: true, - Out: io.Discard, } if err := client.Run(context.Background()); err != nil { t.Fatal(err) @@ -122,7 +121,7 @@ func TestRunEmitsJoinAndPostJoinEvents(t *testing.T) { for { line, err := reader.ReadString('\n') if err != nil { - serverErr <- err + serverErr <- nil return } line = strings.TrimRight(line, "\r\n") @@ -133,7 +132,7 @@ func TestRunEmitsJoinAndPostJoinEvents(t *testing.T) { _, _ = conn.Write([]byte(":viewer!viewer@viewer.tmi.twitch.tv JOIN #streamer\r\n")) _, _ = conn.Write([]byte(":someone!someone@someone.tmi.twitch.tv PRIVMSG #streamer :hello\r\n")) _, _ = conn.Write([]byte("PING :tmi.twitch.tv\r\n")) - case "PONG :tmi.twitch.tv": + case "QUIT :parasocial shutting down": serverErr <- nil return } @@ -151,7 +150,6 @@ func TestRunEmitsJoinAndPostJoinEvents(t *testing.T) { Events: func(event Event) { events <- event }, - Out: io.Discard, } done := make(chan error, 1) @@ -203,6 +201,12 @@ func TestAuthFailureReturnsError(t *testing.T) { return } defer conn.Close() + reader := bufio.NewReader(conn) + for range 3 { + if _, err := reader.ReadString('\n'); err != nil { + return + } + } _, _ = conn.Write([]byte(":tmi.twitch.tv NOTICE * :Login authentication failed\r\n")) }() @@ -211,7 +215,6 @@ func TestAuthFailureReturnsError(t *testing.T) { Login: "viewer", Token: "bad-token", Streamer: "streamer", - Out: io.Discard, } err = client.Run(context.Background()) if err == nil { diff --git a/internal/irc/manager.go b/internal/irc/manager.go index e3eb21a..ae93b2b 100644 --- a/internal/irc/manager.go +++ b/internal/irc/manager.go @@ -2,22 +2,13 @@ package irc import ( "context" - "io" "net" "strings" "sync" ) -// Target identifies one streamer channel that should have an active IRC connection. -type Target struct { - Login string -} - // Manager reconciles the active IRC connections against the desired watched set. type Manager struct { - Addr string - Debug bool - Out io.Writer Events EventSink DialContext func(context.Context, string, string) (net.Conn, error) RunClient func(context.Context, *Client) error @@ -32,7 +23,7 @@ type managedClient struct { } // Sync applies the desired IRC target set using the provided viewer credentials. -func (m *Manager) Sync(ctx context.Context, viewerLogin, accessToken string, targets []Target) { +func (m *Manager) Sync(ctx context.Context, viewerLogin, accessToken string, targets []string) { orderedTargets, desired := normalizeTargets(targets) m.mu.Lock() @@ -82,12 +73,9 @@ func (m *Manager) runClient(ctx context.Context, viewerLogin, accessToken, strea m.emit(Event{Streamer: streamer, State: StatePending}) _ = run(ctx, &Client{ - Addr: m.addr(), Login: viewerLogin, Token: accessToken, Streamer: streamer, - Debug: m.Debug, - Out: m.Out, Events: m.Events, DialContext: m.DialContext, }) @@ -111,13 +99,6 @@ func (m *Manager) Close() { m.wg.Wait() } -func (m *Manager) addr() string { - if m.Addr != "" { - return m.Addr - } - return DefaultAddr -} - func (m *Manager) emit(event Event) { if m.Events == nil { return @@ -125,11 +106,11 @@ func (m *Manager) emit(event Event) { m.Events(event) } -func normalizeTargets(targets []Target) ([]string, map[string]struct{}) { +func normalizeTargets(targets []string) ([]string, map[string]struct{}) { ordered := make([]string, 0, 2) normalized := make(map[string]struct{}, len(targets)) for _, target := range targets { - login := strings.ToLower(strings.TrimSpace(target.Login)) + login := strings.ToLower(strings.TrimSpace(target)) if login == "" { continue } diff --git a/internal/irc/manager_test.go b/internal/irc/manager_test.go index 834ed82..2ca133f 100644 --- a/internal/irc/manager_test.go +++ b/internal/irc/manager_test.go @@ -25,11 +25,7 @@ func TestManagerSyncStartsAtMostTwoTargets(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - manager.Sync(ctx, "viewer", "token", []Target{ - {Login: "alpha"}, - {Login: "beta"}, - {Login: "gamma"}, - }) + manager.Sync(ctx, "viewer", "token", []string{"alpha", "beta", "gamma"}) assertStartSet(t, started, "alpha", "beta") assertNoStart(t, started) @@ -50,10 +46,10 @@ func TestManagerSyncKeepsExistingTargetsConnected(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}}) + manager.Sync(ctx, "viewer", "token", []string{"alpha"}) assertStartSet(t, started, "alpha") - manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}}) + manager.Sync(ctx, "viewer", "token", []string{"alpha"}) assertNoStart(t, started) } @@ -74,10 +70,10 @@ func TestManagerSyncCancelsRemovedTargets(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}, {Login: "beta"}}) + manager.Sync(ctx, "viewer", "token", []string{"alpha", "beta"}) assertStartSet(t, started, "alpha", "beta") - manager.Sync(ctx, "viewer", "token", []Target{{Login: "beta"}, {Login: "gamma"}}) + manager.Sync(ctx, "viewer", "token", []string{"beta", "gamma"}) assertStartSet(t, started, "gamma") assertCancels(t, canceled, "alpha") assertNoStart(t, started) @@ -109,7 +105,7 @@ func TestManagerSyncRestartsTargetAfterFailure(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}}) + manager.Sync(ctx, "viewer", "token", []string{"alpha"}) assertStartSet(t, started, "alpha") waitFor(t, func() bool { @@ -119,7 +115,7 @@ func TestManagerSyncRestartsTargetAfterFailure(t *testing.T) { return !ok }) - manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}}) + manager.Sync(ctx, "viewer", "token", []string{"alpha"}) assertStartSet(t, started, "alpha") } @@ -138,7 +134,7 @@ func TestManagerEmitsPendingAndDisconnectedEvents(t *testing.T) { } ctx, cancel := context.WithCancel(context.Background()) - manager.Sync(ctx, "viewer", "token", []Target{{Login: "alpha"}}) + manager.Sync(ctx, "viewer", "token", []string{"alpha"}) cancel() manager.Close() diff --git a/internal/miner/manager.go b/internal/miner/manager.go index 3e6e522..fb867d9 100644 --- a/internal/miner/manager.go +++ b/internal/miner/manager.go @@ -52,12 +52,11 @@ const ( // 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 + Login string + Watching bool + Reason WatchReason + WatchedMinutes int + WatchStreak *int } // Manager owns background channel points mining state for the current authenticated user. @@ -89,7 +88,6 @@ type streamerState struct { ChannelPoints int SpadeURL string Metadata *twitch.StreamMetadata - LastWatchAt time.Time WatchStreak *int WatchStreakMissing bool @@ -98,10 +96,7 @@ type streamerState struct { CurrentWatchReason WatchReason OnlineAt time.Time OfflineAt time.Time - PendingStreamUpAt time.Time - CurrentBroadcastID string - seeded bool seeding bool refreshing bool } @@ -146,13 +141,9 @@ func (m *Manager) Sync(ctx context.Context, state *auth.State, viewer *twitch.Vi return } - type seedTarget struct { - configLogin string - } - var ( channelIDs []string - seedList []seedTarget + seedList []string statuses []StatusEntry ) @@ -176,7 +167,7 @@ func (m *Manager) Sync(ctx context.Context, state *auth.State, viewer *twitch.Vi Login: entry.Login, ChannelID: entry.ChannelID, } - seedList = append(seedList, seedTarget{configLogin: entry.ConfigLogin}) + seedList = append(seedList, entry.ConfigLogin) } current.ConfigLogin = entry.ConfigLogin @@ -203,8 +194,8 @@ func (m *Manager) Sync(ctx context.Context, state *auth.State, viewer *twitch.Vi m.emitStatuses(statuses) _ = m.pubsub.Sync(ctx, viewer.ID, state.AccessToken, channelIDs) - for _, target := range seedList { - m.scheduleSeed(target.configLogin) + for _, configLogin := range seedList { + m.scheduleSeed(configLogin) } } @@ -256,7 +247,6 @@ func (m *Manager) scheduleSeed(configLogin string) { m.mu.Lock() if current, ok := m.entries[configLogin]; ok && current.ChannelID == channelID { current.ChannelPoints = channelPoints.Balance - current.seeded = true } m.mu.Unlock() m.logf(login, "seeded channel points balance: %d", channelPoints.Balance) @@ -312,7 +302,6 @@ 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() @@ -378,7 +367,6 @@ 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 = m.now() state.Watched += time.Minute if candidate.reason == WatchReasonStreak && state.WatchStreakMissing { state.WatchStreakWatched += time.Minute @@ -475,7 +463,6 @@ func (m *Manager) updateWatchReasonsLocked(candidates []watchCandidate) []Status 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) } @@ -487,7 +474,6 @@ func (m *Manager) markOnlineConfirmedLocked(state *streamerState, now time.Time, state.WatchStreakWatched = 0 state.SpadeURL = "" state.Metadata = nil - state.CurrentBroadcastID = "" } func (m *Manager) markOfflineLocked(state *streamerState, now time.Time) { @@ -497,8 +483,6 @@ func (m *Manager) markOfflineLocked(state *streamerState, now time.Time) { state.Metadata = nil state.Watched = 0 state.CurrentWatchReason = "" - state.PendingStreamUpAt = time.Time{} - state.CurrentBroadcastID = "" } func (m *Manager) emitStatusForConfig(configLogin string) { @@ -534,12 +518,11 @@ func (m *Manager) emitStatuses(statuses []StatusEntry) { 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), + Login: state.Login, + Watching: state.Live && state.CurrentWatchReason != "", + Reason: state.CurrentWatchReason, + WatchedMinutes: int(state.Watched / time.Minute), + WatchStreak: cloneInt(state.WatchStreak), } } @@ -583,12 +566,6 @@ func (m *Manager) handlePubSubEvent(event Event) { m.logf(state.Login, "claim failed: %v", err) } } - case "stream-up": - m.mu.Lock() - if current, ok := m.entries[configLogin]; ok { - current.PendingStreamUpAt = m.now() - } - m.mu.Unlock() case "stream-down": var statuses []StatusEntry m.mu.Lock() diff --git a/internal/miner/manager_test.go b/internal/miner/manager_test.go index 53f17e7..5450901 100644 --- a/internal/miner/manager_test.go +++ b/internal/miner/manager_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "reflect" + "sync" "testing" "time" @@ -12,6 +13,7 @@ import ( ) type fakeService struct { + mu sync.Mutex channelPoints map[string]*twitch.ChannelPointsContext metadata map[string]*twitch.StreamMetadata spadeURLs map[string]string @@ -20,9 +22,8 @@ type fakeService struct { minuteErr map[string]error claimErr map[string]error - claimed []string - watched []string - refreshed []string + claimed []string + watched []string } func (f *fakeService) LoadChannelPointsContext(_ context.Context, login string) (*twitch.ChannelPointsContext, error) { @@ -43,6 +44,8 @@ func (f *fakeService) WatchStreak(_ context.Context, login string) (*int, error) } func (f *fakeService) ClaimCommunityPoints(_ context.Context, channelID, claimID string) error { + f.mu.Lock() + defer f.mu.Unlock() f.claimed = append(f.claimed, channelID+":"+claimID) if err, ok := f.claimErr[channelID+":"+claimID]; ok { return err @@ -51,7 +54,6 @@ func (f *fakeService) ClaimCommunityPoints(_ context.Context, channelID, claimID } 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 } @@ -73,11 +75,15 @@ func (f *fakeService) PlaybackAccessToken(_ context.Context, login string) (*twi } func (f *fakeService) TouchPlayback(_ context.Context, login string, _ *twitch.PlaybackToken) error { + f.mu.Lock() + defer f.mu.Unlock() f.watched = append(f.watched, "touch:"+login) return nil } func (f *fakeService) SendMinuteWatched(_ context.Context, spadeURL string, _ twitch.MinuteWatchedPayload) error { + f.mu.Lock() + defer f.mu.Unlock() f.watched = append(f.watched, "post:"+spadeURL) if err, ok := f.minuteErr[spadeURL]; ok { return err @@ -85,6 +91,18 @@ func (f *fakeService) SendMinuteWatched(_ context.Context, spadeURL string, _ tw return nil } +func (f *fakeService) claimedCalls() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.claimed...) +} + +func (f *fakeService) watchedCalls() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.watched...) +} + type fakePubSub struct { syncCalls [][]string } @@ -120,7 +138,7 @@ func TestManagerSyncSeedsAndClaims(t *testing.T) { 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 != "" + return len(service.claimedCalls()) == 1 && manager.entries["alpha"].ChannelPoints == 250 && manager.entries["alpha"].SpadeURL != "" }) if got, want := pubsub.syncCalls, [][]string{{"1"}}; !reflect.DeepEqual(got, want) { @@ -146,7 +164,7 @@ func TestManagerWatchOnceUsesTopTwoLiveStreamers(t *testing.T) { } manager := NewManager(context.Background(), service, &fakePubSub{}, nil) defer manager.Close() - manager.sleep = cancelSleep + manager.watchStarted = true 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}, @@ -163,7 +181,7 @@ func TestManagerWatchOnceUsesTopTwoLiveStreamers(t *testing.T) { if err := manager.watchOnce(context.Background()); err != nil { t.Fatal(err) } - if got, want := service.watched, []string{ + if got, want := service.watchedCalls(), []string{ "touch:alpha_live", "post:https://spade.test/alpha", "touch:beta_live", @@ -189,7 +207,7 @@ func TestManagerWatchOncePrioritizesMissingWatchStreaks(t *testing.T) { t.Fatal(err) } - if got, want := service.watched, []string{ + if got, want := service.watchedCalls(), []string{ "touch:beta_live", "post:https://spade.test/beta_live", "touch:gamma_live", @@ -215,7 +233,7 @@ func TestManagerWatchOnceFillsWithPointsCandidates(t *testing.T) { t.Fatal(err) } - if got, want := service.watched, []string{ + if got, want := service.watchedCalls(), []string{ "touch:beta_live", "post:https://spade.test/beta_live", "touch:alpha_live", @@ -335,7 +353,7 @@ func TestHandlePubSubWatchStreakCompletesMaintenance(t *testing.T) { CurrentWatchReason: WatchReasonStreak, } - manager.handlePubSubEvent(Event{MessageType: "points-earned", ChannelID: "1", Balance: 555, ReasonCode: "WATCH_STREAK", TotalPoints: 450, Timestamp: "t1", Topic: "community-points-user-v1"}) + manager.handlePubSubEvent(Event{MessageType: "points-earned", ChannelID: "1", Balance: 555, ReasonCode: "WATCH_STREAK", Timestamp: "t1", Topic: "community-points-user-v1"}) waitFor(t, func() bool { manager.mu.Lock() @@ -434,26 +452,6 @@ func TestManagerSyncPollingConfirmedOnlineResetsStreakState(t *testing.T) { } } -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) @@ -468,7 +466,7 @@ func TestHandlePubSubEventUpdatesBalanceAndClaims(t *testing.T) { 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) { + if got, want := service.claimedCalls(), []string{"1:claim-2"}; !reflect.DeepEqual(got, want) { t.Fatalf("claimed = %#v, want %#v", got, want) } } diff --git a/internal/miner/pubsub.go b/internal/miner/pubsub.go index a96b898..f723a42 100644 --- a/internal/miner/pubsub.go +++ b/internal/miner/pubsub.go @@ -6,7 +6,8 @@ import ( "errors" "fmt" "net/http" - "sort" + "slices" + "strings" "sync" "time" @@ -29,7 +30,6 @@ type Event struct { Balance int ClaimID string ReasonCode string - TotalPoints int } func (e Event) key() string { @@ -45,7 +45,6 @@ type Client struct { dialer *websocket.Dialer mu sync.Mutex conn *websocket.Conn - viewerID string token string topics []string updateCh chan struct{} @@ -81,12 +80,11 @@ func (c *Client) Sync(_ context.Context, viewerID, accessToken string, channelID } topics = append(topics, "video-playback-by-id."+channelID) } - sort.Strings(topics) + slices.Sort(topics) c.mu.Lock() - c.viewerID = viewerID c.token = accessToken - changed := !equalStrings(c.topics, topics) + changed := !slices.Equal(c.topics, topics) c.topics = topics conn := c.conn c.mu.Unlock() @@ -258,15 +256,13 @@ func (c *Client) snapshot() ([]string, string) { 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 { + Type string `json:"type"` + Data *struct { Topic string `json:"topic"` Message string `json:"message"` } `json:"data"` @@ -275,7 +271,7 @@ func parseFrame(message []byte) (*frame, error) { return nil, err } - result := &frame{Type: envelope.Type, Error: envelope.Error} + result := &frame{Type: envelope.Type} if envelope.Type != "MESSAGE" || envelope.Data == nil { return result, nil } @@ -294,14 +290,18 @@ func parseFrame(message []byte) (*frame, error) { ChannelID string `json:"channel_id"` } `json:"balance"` PointGain *struct { - ReasonCode string `json:"reason_code"` - TotalPoints int `json:"total_points"` + ReasonCode string `json:"reason_code"` } `json:"point_gain"` } `json:"data"` } if err := json.Unmarshal([]byte(envelope.Data.Message), &payload); err != nil { return nil, err } + topic := envelope.Data.Topic + topicName, topicChannelID := topic, "" + if i := strings.LastIndexByte(topic, '.'); i >= 0 { + topicName, topicChannelID = topic[:i], topic[i+1:] + } channelID := payload.Data.ChannelID if payload.Data.Claim != nil && payload.Data.Claim.ChannelID != "" { @@ -311,11 +311,11 @@ func parseFrame(message []byte) (*frame, error) { channelID = payload.Data.Balance.ChannelID } if channelID == "" { - channelID = topicSuffix(envelope.Data.Topic) + channelID = topicChannelID } result.Event = Event{ - Topic: topicPrefix(envelope.Data.Topic), + Topic: topicName, MessageType: payload.Type, ChannelID: channelID, Timestamp: payload.Data.Timestamp, @@ -328,37 +328,6 @@ func parseFrame(message []byte) (*frame, error) { } if payload.Data.PointGain != nil { result.Event.ReasonCode = payload.Data.PointGain.ReasonCode - result.Event.TotalPoints = payload.Data.PointGain.TotalPoints } 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 index ad78f8d..6dbbfb0 100644 --- a/internal/miner/pubsub_test.go +++ b/internal/miner/pubsub_test.go @@ -21,7 +21,7 @@ func TestParseFramePointsEarnedPointGain(t *testing.T) { if err != nil { t.Fatal(err) } - if frame.Event.ReasonCode != "WATCH_STREAK" || frame.Event.TotalPoints != 450 { + if frame.Event.ReasonCode != "WATCH_STREAK" { t.Fatalf("event = %#v", frame.Event) } } diff --git a/internal/tui/messages.go b/internal/tui/messages.go index 228f284..2c017c1 100644 --- a/internal/tui/messages.go +++ b/internal/tui/messages.go @@ -2,18 +2,11 @@ package tui import ( "parasocial/internal/auth" + "parasocial/internal/irc" + "parasocial/internal/miner" "parasocial/internal/twitch" ) -// IRCState describes the current IRC join lifecycle for one streamer row. -type IRCState string - -const ( - IRCPending IRCState = "pending" - IRCJoined IRCState = "joined" - IRCDisconnected IRCState = "disconnected" -) - // AuthUpdate carries one incremental auth log line or completion result into the TUI. type AuthUpdate struct { Line string @@ -24,36 +17,14 @@ type AuthUpdate struct { // StreamerUpdate carries one streamer resolution update into the TUI. type StreamerUpdate struct { - Viewer *twitch.Viewer - Entry *twitch.StreamerEntry - IRC *IRCUpdate - Miner *MinerUpdate - Index int - Err error - Done bool -} - -// IRCUpdate carries one IRC connection state or log line into the TUI. -type IRCUpdate struct { - Login string - State IRCState - Line string -} - -// MinerUpdate carries one miner log line into the TUI. -type MinerUpdate struct { - 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 + Viewer *twitch.Viewer + Entry *twitch.StreamerEntry + IRC *irc.Event + MinerLog *miner.LogEntry + MinerStatus *miner.StatusEntry + Index int + Err error + Done bool } type authStartedMsg struct { diff --git a/internal/tui/model.go b/internal/tui/model.go index 091f107..96e8926 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -8,6 +8,8 @@ import ( tea "github.com/charmbracelet/bubbletea" "parasocial/internal/auth" + "parasocial/internal/irc" + "parasocial/internal/miner" "parasocial/internal/twitch" ) @@ -61,7 +63,7 @@ type Model struct { focus panelFocus ircDetails map[string]ircDetail minerDetails map[string][]string - minerStatuses map[string]MinerStatus + minerStatuses map[string]miner.StatusEntry authViewport viewport.Model ircViewport viewport.Model minerViewport viewport.Model @@ -87,13 +89,13 @@ func New(options Options) Model { mode: authView, ircDetails: make(map[string]ircDetail), minerDetails: make(map[string][]string), - minerStatuses: make(map[string]MinerStatus), + minerStatuses: make(map[string]miner.StatusEntry), width: defaultViewWidth, height: 24, focus: focusStreamers, - authViewport: newAuthViewport(contentWidth(defaultViewWidth), authViewportHeight(24)), - ircViewport: newIRCViewport(detailViewportWidth(defaultViewWidth), detailViewportHeight(24, "")), - minerViewport: newMinerViewport( + authViewport: viewport.New(contentWidth(defaultViewWidth), authViewportHeight(24)), + ircViewport: viewport.New(detailViewportWidth(defaultViewWidth), detailViewportHeight(24, "")), + minerViewport: viewport.New( detailViewportWidth(defaultViewWidth), detailViewportHeight(24, ""), ), @@ -206,7 +208,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.minerStatuses = make(map[string]miner.StatusEntry) m.selectedConfig = "" m.focus = focusStreamers m.ensureSelection() @@ -233,8 +235,17 @@ func (m Model) updateStreamer(msg StreamerUpdate) (tea.Model, tea.Cmd) { if msg.IRC != nil { m.applyIRCUpdate(*msg.IRC) } - if msg.Miner != nil { - m.applyMinerUpdate(*msg.Miner) + if update := msg.MinerLog; update != nil { + if login := normalizeKey(update.Login); login != "" { + if line := strings.TrimSpace(update.Line); line != "" { + m.minerDetails[login] = appendCappedHistory(m.minerDetails[login], line) + } + } + } + if update := msg.MinerStatus; update != nil { + if login := normalizeKey(update.Login); login != "" { + m.minerStatuses[login] = *update + } } if msg.Err != nil { m.resolveErr = msg.Err @@ -251,17 +262,17 @@ func (m Model) updateStreamer(msg StreamerUpdate) (tea.Model, tea.Cmd) { return m, nil } -func (m *Model) applyIRCUpdate(update IRCUpdate) { - login := normalizeKey(update.Login) +func (m *Model) applyIRCUpdate(update irc.Event) { + login := normalizeKey(update.Streamer) if login == "" { return } detail := m.ircDetails[login] switch update.State { - case IRCJoined: + case irc.StateJoined: detail.joined = true - case IRCPending, IRCDisconnected: + case irc.StatePending, irc.StateDisconnected: detail.joined = false } if message, ok := formatIRCChatLine(update.Line); ok { @@ -270,20 +281,6 @@ func (m *Model) applyIRCUpdate(update IRCUpdate) { m.ircDetails[login] = detail } -func (m *Model) applyMinerUpdate(update MinerUpdate) { - login := normalizeKey(update.Login) - if login == "" { - return - } - 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() { entries := m.orderedStreamers() if len(entries) == 0 { @@ -482,14 +479,6 @@ func (m Model) visibleDetailTab() detailTab { } } -func (m Model) isStreamersFocused() bool { - return m.focus == focusStreamers -} - -func (m Model) isRightPanelFocused() bool { - return m.focus == focusInfo || m.focus == focusIRC || m.focus == focusMiner -} - func isActive(entry twitch.StreamerEntry) bool { return entry.Status == twitch.StreamerReady && entry.Live } diff --git a/internal/tui/model_test.go b/internal/tui/model_test.go index 28521c3..40d16c2 100644 --- a/internal/tui/model_test.go +++ b/internal/tui/model_test.go @@ -8,6 +8,8 @@ import ( tea "github.com/charmbracelet/bubbletea" "parasocial/internal/auth" + "parasocial/internal/irc" + "parasocial/internal/miner" "parasocial/internal/twitch" ) @@ -59,7 +61,7 @@ func TestAuthUpdateAppendsLogLine(t *testing.T) { func TestAuthSuccessSwitchesToStreamerViewAndStartsResolution(t *testing.T) { runtime := &fakeModelRuntime{} - state := &auth.State{Login: "viewer", UserID: "7"} + state := &auth.State{Login: "viewer"} updated, cmd := New(Options{Streamers: []string{"alpha"}, runtime: runtime}).Update(AuthUpdate{State: state, Done: true}) if cmd == nil { t.Fatal("expected streamer resolution command") @@ -87,7 +89,7 @@ func TestInitStartsAuthOrResolution(t *testing.T) { t.Fatal("expected auth runtime to start") } - state := &auth.State{Login: "viewer", UserID: "7"} + state := &auth.State{Login: "viewer"} runtime = &fakeModelRuntime{} if _, ok := New(Options{Streamers: []string{"alpha"}, AuthState: state, runtime: runtime}).Init()().(streamerStartedMsg); !ok { t.Fatal("authenticated Init() did not start streamer resolution") @@ -177,13 +179,11 @@ func TestInfoTabShowsWatchStreakMinerStatus(t *testing.T) { Status: twitch.StreamerReady, WatchStreak: &streak, }) - updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ - Login: "alpha_live", - Status: &MinerStatus{ - Watching: true, - Reason: "watchstreak", - WatchedMinutes: 4, - }, + updated, _ := model.Update(StreamerUpdate{MinerStatus: &miner.StatusEntry{ + Login: "alpha_live", + Watching: true, + Reason: miner.WatchReasonStreak, + WatchedMinutes: 4, }}) next := updated.(Model) entry, ok := next.selectedEntry() @@ -206,14 +206,12 @@ func TestInfoTabShowsPointsMinerStatus(t *testing.T) { Status: twitch.StreamerReady, WatchStreak: &streak, }) - updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ - Login: "alpha_live", - Status: &MinerStatus{ - Watching: true, - Reason: "points", - WatchedMinutes: 6, - WatchStreak: &minerStreak, - }, + updated, _ := model.Update(StreamerUpdate{MinerStatus: &miner.StatusEntry{ + Login: "alpha_live", + Watching: true, + Reason: miner.WatchReasonPoints, + WatchedMinutes: 6, + WatchStreak: &minerStreak, }}) next := updated.(Model) entry, ok := next.selectedEntry() @@ -233,12 +231,10 @@ func TestInfoTabOmitsMinerStatusWhenInactiveOrUnwatched(t *testing.T) { Status: twitch.StreamerReady, WatchStreak: &streak, }) - updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ - Login: "alpha_live", - Status: &MinerStatus{ - Watching: true, - Reason: "points", - }, + updated, _ := model.Update(StreamerUpdate{MinerStatus: &miner.StatusEntry{ + Login: "alpha_live", + Watching: true, + Reason: miner.WatchReasonPoints, }}) next := updated.(Model) entry, ok := next.selectedEntry() @@ -427,16 +423,16 @@ func TestIRCUpdatesShowJoinedStatusAndFormattedMessages(t *testing.T) { Login: "alpha_live", Live: true, Status: twitch.StreamerReady, - }).Update(StreamerUpdate{IRC: &IRCUpdate{Login: "alpha_live", State: IRCJoined}}) + }).Update(StreamerUpdate{IRC: &irc.Event{Streamer: "alpha_live", State: irc.StateJoined}}) next := updated.(Model) if !next.ircDetails["alpha_live"].joined { t.Fatal("expected joined detail") } - updated, _ = next.Update(StreamerUpdate{IRC: &IRCUpdate{ - Login: "alpha_live", - Line: ":someone!someone@someone.tmi.twitch.tv PRIVMSG #alpha_live :hello there", + updated, _ = next.Update(StreamerUpdate{IRC: &irc.Event{ + Streamer: "alpha_live", + Line: ":someone!someone@someone.tmi.twitch.tv PRIVMSG #alpha_live :hello there", }}) next = updated.(Model) next.focus = focusIRC @@ -482,11 +478,11 @@ func TestIRCUpdatesKeepOnlyLast50ChatMessages(t *testing.T) { Status: twitch.StreamerReady, }) - updated, _ := model.Update(StreamerUpdate{IRC: &IRCUpdate{Login: "alpha_live", State: IRCJoined}}) + updated, _ := model.Update(StreamerUpdate{IRC: &irc.Event{Streamer: "alpha_live", State: irc.StateJoined}}) next := updated.(Model) for i := 1; i <= maxIRCMessageHistory+5; i++ { line := fmt.Sprintf(":user%d!user%d@user%d.tmi.twitch.tv PRIVMSG #alpha_live :message %d", i, i, i, i) - updated, _ = next.Update(StreamerUpdate{IRC: &IRCUpdate{Login: "alpha_live", Line: line}}) + updated, _ = next.Update(StreamerUpdate{IRC: &irc.Event{Streamer: "alpha_live", Line: line}}) next = updated.(Model) } @@ -512,7 +508,7 @@ func TestMinerUpdatesRenderAndKeepOnlyLast50Messages(t *testing.T) { next := model for i := 1; i <= maxIRCMessageHistory+5; i++ { - updated, _ := next.Update(StreamerUpdate{Miner: &MinerUpdate{ + updated, _ := next.Update(StreamerUpdate{MinerLog: &miner.LogEntry{ Login: "alpha_live", Line: fmt.Sprintf("miner message %d", i), }}) @@ -541,7 +537,7 @@ func TestMinerTabShowsHistoryForOfflineStreamer(t *testing.T) { Status: twitch.StreamerReady, }) - updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ + updated, _ := model.Update(StreamerUpdate{MinerLog: &miner.LogEntry{ Login: "alpha_live", Line: "pubsub stream down", }}) @@ -560,9 +556,9 @@ func TestIRCUpdatesIgnoreNonChatProtocolLines(t *testing.T) { Status: twitch.StreamerReady, }) - updated, _ := model.Update(StreamerUpdate{IRC: &IRCUpdate{ - Login: "alpha_live", - Line: "Joined #alpha_live as viewer", + updated, _ := model.Update(StreamerUpdate{IRC: &irc.Event{ + Streamer: "alpha_live", + Line: "Joined #alpha_live as viewer", }}) next := updated.(Model) if len(next.ircDetails["alpha_live"].messages) != 0 { @@ -588,9 +584,9 @@ func TestIRCViewportAutoScrollsAtBottom(t *testing.T) { t.Fatal("expected viewport to start at bottom") } - updated, _ := model.Update(StreamerUpdate{IRC: &IRCUpdate{ - Login: "alpha_live", - Line: ":late!late@late.tmi.twitch.tv PRIVMSG #alpha_live :newest", + updated, _ := model.Update(StreamerUpdate{IRC: &irc.Event{ + Streamer: "alpha_live", + Line: ":late!late@late.tmi.twitch.tv PRIVMSG #alpha_live :newest", }}) next := updated.(Model) if !next.ircViewport.AtBottom() { @@ -613,9 +609,9 @@ func TestIRCViewportPreservesManualScrollPosition(t *testing.T) { model.syncIRCViewport(true) model.ircViewport.GotoTop() - updated, _ := model.Update(StreamerUpdate{IRC: &IRCUpdate{ - Login: "alpha_live", - Line: ":late!late@late.tmi.twitch.tv PRIVMSG #alpha_live :newest", + updated, _ := model.Update(StreamerUpdate{IRC: &irc.Event{ + Streamer: "alpha_live", + Line: ":late!late@late.tmi.twitch.tv PRIVMSG #alpha_live :newest", }}) next := updated.(Model) if next.ircViewport.YOffset != 0 { @@ -641,7 +637,7 @@ func TestMinerViewportAutoScrollsAtBottom(t *testing.T) { t.Fatal("expected miner viewport to start at bottom") } - updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ + updated, _ := model.Update(StreamerUpdate{MinerLog: &miner.LogEntry{ Login: "alpha_live", Line: "newest miner event", }}) @@ -663,7 +659,7 @@ func TestMinerViewportPreservesManualScrollPosition(t *testing.T) { model.syncMinerViewport(true) model.minerViewport.GotoTop() - updated, _ := model.Update(StreamerUpdate{Miner: &MinerUpdate{ + updated, _ := model.Update(StreamerUpdate{MinerLog: &miner.LogEntry{ Login: "alpha_live", Line: "newest miner event", }}) diff --git a/internal/tui/runtime.go b/internal/tui/runtime.go index 252d00e..116c440 100644 --- a/internal/tui/runtime.go +++ b/internal/tui/runtime.go @@ -4,9 +4,7 @@ import ( "context" "errors" "fmt" - "io" "net/http" - "os" "strings" "time" @@ -25,23 +23,16 @@ const streamerRefreshInterval = 5 * time.Minute type Options struct { Streamers []string AuthState *auth.State - AuthPath string - Input io.Reader - Output io.Writer runtime modelRuntime initialStreamers []twitch.StreamerEntry - httpClient *http.Client } type runtime struct { - ctx context.Context - logins []string - httpClient *http.Client - authClient *auth.Client - authPath string - refreshEvery time.Duration - sleep func(context.Context, time.Duration) error + ctx context.Context + logins []string + httpClient *http.Client + authClient *auth.Client } // Run starts the Bubble Tea UI and keeps application orchestration inside the TUI package. @@ -49,11 +40,17 @@ func Run(ctx context.Context, options Options) error { ctx, cancel := context.WithCancel(ctx) defer cancel() - rt := newRuntime(ctx, options) + httpClient := &http.Client{Timeout: 30 * time.Second} + rt := &runtime{ + ctx: ctx, + logins: append([]string(nil), options.Streamers...), + httpClient: httpClient, + authClient: auth.NewClient(httpClient), + } state := options.AuthState if state == nil { var err error - state, err = rt.reuseAuth() + state, err = rt.authClient.ReuseAuth(rt.ctx, auth.DefaultPath()) if err != nil { return err } @@ -62,54 +59,19 @@ func Run(ctx context.Context, options Options) error { options.AuthState = state options.runtime = rt - input := options.Input - if input == nil { - input = os.Stdin - } - output := options.Output - if output == nil { - output = os.Stdout - } - program := tea.NewProgram( New(options), tea.WithContext(ctx), - tea.WithInput(input), - tea.WithOutput(output), ) _, err := program.Run() return err } -func newRuntime(ctx context.Context, options Options) *runtime { - httpClient := options.httpClient - if httpClient == nil { - httpClient = &http.Client{Timeout: 30 * time.Second} - } - authPath := options.AuthPath - if authPath == "" { - authPath = auth.DefaultPath() - } - return &runtime{ - ctx: ctx, - logins: append([]string(nil), options.Streamers...), - httpClient: httpClient, - authClient: auth.NewClient(httpClient), - authPath: authPath, - refreshEvery: streamerRefreshInterval, - sleep: sleepContext, - } -} - -func (r *runtime) reuseAuth() (*auth.State, error) { - return r.authClient.ReuseAuth(r.ctx, r.authPath) -} - func (r *runtime) startAuth(ch chan<- AuthUpdate) { go func() { defer close(ch) - state, err := r.authClient.EnsureAuth(r.ctx, r.authPath, func(line string) { + state, err := r.authClient.EnsureAuth(r.ctx, auth.DefaultPath(), func(line string) { ch <- AuthUpdate{Line: strings.TrimRight(line, "\n")} }) if err != nil { @@ -128,20 +90,40 @@ func (r *runtime) startAuth(ch chan<- AuthUpdate) { func (r *runtime) startResolve(state *auth.State, ch chan<- StreamerUpdate) { go func() { defer close(ch) - ircManager := newIRCManager(ch) + ircManager := &irc.Manager{Events: func(event irc.Event) { + ch <- StreamerUpdate{IRC: &event} + }} defer ircManager.Close() - service, err := newTwitchService(r.httpClient, state) - if err != nil { - ch <- StreamerUpdate{Err: err, Done: true} + client := &gql.Client{ + HTTPClient: r.httpClient, + Session: gql.Session{ + AccessToken: state.AccessToken, + ClientID: auth.ClientID, + DeviceID: state.DeviceID, + UserAgent: auth.TVUserAgent, + }, + } + if err := client.Validate(); err != nil { + ch <- StreamerUpdate{Err: fmt.Errorf("configure graphql session: %w", err), Done: true} return } - minerManager := newMinerManager(r.ctx, service, ch) + service := &twitch.Service{ + GQL: client, + HTTPClient: r.httpClient, + Session: client.Session, + } + minerManager := miner.NewManager(r.ctx, service, nil, func(entry miner.LogEntry) { + ch <- StreamerUpdate{MinerLog: &entry} + }) + minerManager.SetStatusSink(func(entry miner.StatusEntry) { + ch <- StreamerUpdate{MinerStatus: &entry} + }) 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 { + }, streamerRefreshInterval, sleepContext); err != nil { if errors.Is(err, context.Canceled) { return } @@ -158,95 +140,16 @@ type streamerService interface { 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 - 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) + Sync(context.Context, string, string, []string) } 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, - Session: gql.Session{ - AccessToken: state.AccessToken, - ClientID: state.ClientID, - DeviceID: state.DeviceID, - UserAgent: auth.TVUserAgent(), - }, - } - if err := client.Validate(); err != nil { - return nil, fmt.Errorf("configure graphql session: %w", err) - } - return &twitch.Service{ - GQL: client, - HTTPClient: httpClient, - Session: client.Session, - }, nil -} - -func newIRCManager(ch chan<- StreamerUpdate) *irc.Manager { - return &irc.Manager{ - Addr: irc.DefaultAddr, - Events: func(event irc.Event) { - ch <- StreamerUpdate{ - IRC: &IRCUpdate{ - Login: event.Streamer, - State: IRCState(event.State), - Line: event.Line, - }, - } - }, - } -} - -func newMinerManager(ctx context.Context, service miner.Service, ch chan<- StreamerUpdate) *miner.Manager { - manager := miner.NewManager(ctx, service, nil, newMinerLogSink(ch)) - manager.SetStatusSink(newMinerStatusSink(ch)) - return manager -} - -func newMinerLogSink(ch chan<- StreamerUpdate) func(miner.LogEntry) { - return func(entry miner.LogEntry) { - ch <- StreamerUpdate{ - Miner: &MinerUpdate{ - Login: entry.Login, - Line: entry.Line, - }, - } - } -} - -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) -} - 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 { @@ -348,24 +251,16 @@ func syncRuntimeState(ctx context.Context, ircSyncer ircSyncer, minerSyncer mine } } -func watchedIRCTargets(entries []twitch.StreamerEntry) []irc.Target { - targets := make([]irc.Target, 0, 2) +func watchedIRCTargets(entries []twitch.StreamerEntry) []string { + targets := make([]string, 0, 2) for _, entry := range entries { if entry.Status != twitch.StreamerReady || !entry.Live || entry.Login == "" { continue } - targets = append(targets, irc.Target{Login: entry.Login}) + targets = append(targets, entry.Login) if len(targets) == 2 { break } } 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 dc870d6..598cb45 100644 --- a/internal/tui/runtime_test.go +++ b/internal/tui/runtime_test.go @@ -8,8 +8,6 @@ import ( "time" "parasocial/internal/auth" - "parasocial/internal/irc" - "parasocial/internal/miner" "parasocial/internal/twitch" ) @@ -73,40 +71,12 @@ 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 } -func (f *fakeIRCSyncer) Sync(_ context.Context, _, _ string, targets []irc.Target) { - logins := make([]string, 0, len(targets)) - for _, target := range targets { - logins = append(logins, target.Login) - } - f.calls = append(f.calls, logins) +func (f *fakeIRCSyncer) Sync(_ context.Context, _, _ string, targets []string) { + f.calls = append(f.calls, append([]string{}, targets...)) } type fakeMinerSyncer struct { @@ -249,23 +219,6 @@ func TestResolveStreamerEntriesRefreshUpdatesWatchedChannels(t *testing.T) { }) } -func TestNewMinerLogSinkBridgesMinerEntriesIntoStreamerUpdates(t *testing.T) { - ch := make(chan StreamerUpdate, 1) - - newMinerLogSink(ch)(miner.LogEntry{ - Login: "alpha_live", - Line: "pubsub points earned: balance=42", - }) - - update := <-ch - if update.Miner == nil { - t.Fatal("expected miner update") - } - if update.Miner.Login != "alpha_live" || update.Miner.Line != "pubsub points earned: balance=42" { - t.Fatalf("miner update = %#v", update.Miner) - } -} - type serviceOption func(*fakeStreamerService) type watchStreakResult struct { diff --git a/internal/tui/styles.go b/internal/tui/styles.go index 9e549a4..3dfd174 100644 --- a/internal/tui/styles.go +++ b/internal/tui/styles.go @@ -1,7 +1,6 @@ package tui import ( - "github.com/charmbracelet/bubbles/viewport" "github.com/charmbracelet/lipgloss" ) @@ -55,21 +54,3 @@ var ( BorderForeground(accentColor). Foreground(titleColor) ) - -func newAuthViewport(width, height int) viewport.Model { - vp := viewport.New(width, height) - vp.Style = lipgloss.NewStyle() - return vp -} - -func newIRCViewport(width, height int) viewport.Model { - vp := viewport.New(width, height) - vp.Style = lipgloss.NewStyle() - return vp -} - -func newMinerViewport(width, height int) viewport.Model { - vp := viewport.New(width, height) - vp.Style = lipgloss.NewStyle() - return vp -} diff --git a/internal/tui/view.go b/internal/tui/view.go index e8c3337..d141b09 100644 --- a/internal/tui/view.go +++ b/internal/tui/view.go @@ -6,6 +6,7 @@ import ( "github.com/charmbracelet/lipgloss" + "parasocial/internal/miner" "parasocial/internal/twitch" ) @@ -39,10 +40,10 @@ func (m Model) renderStreamerView() string { header := m.renderDashboardHeader() leftStyle := panelStyle rightStyle := panelStyle - if m.isStreamersFocused() { + if m.focus == focusStreamers { leftStyle = focusedPanelStyle } - if m.isRightPanelFocused() { + if m.focus != focusStreamers { rightStyle = focusedPanelStyle } @@ -91,9 +92,9 @@ func (m Model) renderDetailPanel() string { var body string switch m.visibleDetailTab() { case ircTab: - body = m.renderIRCTab() + body = m.ircViewport.View() case minerTab: - body = m.renderMinerTab() + body = m.minerViewport.View() default: body = m.renderInfoTab(entry) } @@ -143,14 +144,6 @@ func (m Model) renderInfoTab(entry twitch.StreamerEntry) string { return strings.Join(lines, "\n") } -func (m Model) renderIRCTab() string { - return m.ircViewport.View() -} - -func (m Model) renderMinerTab() string { - return m.minerViewport.View() -} - func (m Model) renderStreamerRows(maxRows int) string { entries := m.visibleStreamers(maxRows) if len(entries) == 0 { @@ -248,9 +241,9 @@ func (m Model) minerWatchingLines(entry twitch.StreamerEntry) []string { } var reason string switch status.Reason { - case string(minerWatchReasonStreak): + case miner.WatchReasonStreak: reason = "Watching for watchstreak" - case string(minerWatchReasonPoints): + case miner.WatchReasonPoints: reason = "Watching for points" default: return nil @@ -261,33 +254,6 @@ func (m Model) minerWatchingLines(entry twitch.StreamerEntry) []string { } } -type minerWatchReason string - -const ( - minerWatchReasonStreak minerWatchReason = "watchstreak" - minerWatchReasonPoints minerWatchReason = "points" -) - -func rawStatus(entry twitch.StreamerEntry) string { - switch { - case entry.Status == twitch.StreamerError: - return "error" - case entry.Status == twitch.StreamerLoading: - return "loading" - case entry.Live: - return "live" - default: - return "offline" - } -} - -func ircSummary(detail ircDetail) string { - if detail.joined { - return "irc joined" - } - return "irc idle" -} - func contentWidth(width int) int { if width <= 0 { return defaultViewWidth diff --git a/internal/twitch/service.go b/internal/twitch/service.go index 9e57f5b..8df0897 100644 --- a/internal/twitch/service.go +++ b/internal/twitch/service.go @@ -74,25 +74,14 @@ type ChannelPointsContext struct { // Game describes the current Twitch category for a live stream. type Game struct { - ID string - Name string - DisplayName string + ID string + Name 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 + BroadcastID string + Game *Game } // MinuteWatchedPayload is the Spade telemetry body Twitch accepts for simulated viewing. @@ -317,20 +306,13 @@ func (s *Service) StreamMetadata(ctx context.Context, login string) (*StreamMeta 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"` + Game *struct { + ID string `json:"id"` + Name string `json:"name"` } `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"` + ID string `json:"id"` } `json:"stream"` } `json:"user"` } @@ -344,25 +326,13 @@ func (s *Service) StreamMetadata(ctx context.Context, login string) (*StreamMeta 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, - } + metadata := &StreamMetadata{BroadcastID: data.User.Stream.ID} if game := data.User.BroadcastSettings.Game; game != nil { metadata.Game = &Game{ - ID: game.ID, - Name: game.Name, - DisplayName: game.DisplayName, + ID: game.ID, + Name: game.Name, } } - 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) } @@ -423,7 +393,7 @@ func (s *Service) TouchPlayback(ctx context.Context, login string, token *Playba if token == nil || token.Signature == "" || token.Value == "" { return errors.New("playback access token is incomplete") } - masterURL := gql.HLSMasterPlaylistURL(login, token.Signature, token.Value) + masterURL := "https://usher.ttvnw.net/api/channel/hls/" + login + ".m3u8?player=twitchweb&allow_source=true&type=any&p=1&sig=" + token.Signature + "&token=" + token.Value variantURL, err := s.fetchPlaylistTarget(ctx, masterURL) if err != nil { return fmt.Errorf("fetch master playlist: %w", err) diff --git a/internal/twitch/service_test.go b/internal/twitch/service_test.go index 04b06cc..c052258 100644 --- a/internal/twitch/service_test.go +++ b/internal/twitch/service_test.go @@ -214,22 +214,19 @@ 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"}]}}}`, + "VideoPlayerStreamInfoOverlayChannel": `{"user":{"broadcastSettings":{"game":{"id":"99","name":"game-name"}},"stream":{"id":"broadcast"}}}`, }} 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" { + if metadata.BroadcastID != "broadcast" { 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) {