Start implementing different event types in Twitch

Author: xaionaro

go.mod

@@ -29,6 +29,10 @@ replace github.com/asticode/go-astiav v0.36.0 => github.com/xaionaro-go/astiav v
 
 replace github.com/bluenviron/mediacommon/v2 v2.0.1-0.20250324151931-b8ce69d15d3d => github.com/xaionaro-go/mediacommon/v2 v2.0.0-20250420012906-03d6d69ac3b7
 
+replace github.com/joeyak/go-twitch-eventsub/v3 => github.com/xaionaro-go/go-twitch-eventsub/v3 v3.0.0-20250713163657-276e5a5b7adc
+
+replace github.com/scorfly/gokick => github.com/xaionaro-go/gokick v0.0.0-20250713175656-f538d6396851
+
 require (
 	github.com/facebookincubator/go-belt v0.0.0-20250308011339-62fb7027b11f
 	github.com/go-git/go-billy/v5 v5.6.2
@@ -51,6 +55,7 @@ require (
 	fyne.io/systray v1.11.0 // indirect
 	github.com/BurntSushi/toml v1.5.0 // indirect
 	github.com/Danny-Dasilva/CycleTLS/cycletls v1.0.26 // indirect
+	github.com/Danny-Dasilva/fhttp v0.0.0-20240217042913-eeeb0b347ce1 // indirect
 	github.com/DataDog/gostackparse v0.7.0 // indirect
 	github.com/MicahParks/jwkset v0.8.0 // indirect
 	github.com/MicahParks/keyfunc/v3 v3.3.10 // indirect
@@ -252,7 +257,6 @@ require (
 	fyne.io/fyne/v2 v2.5.5
 	github.com/AgustinSRG/go-child-process-manager v1.0.1
 	github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802
-	github.com/Danny-Dasilva/fhttp v0.0.0-20240217042913-eeeb0b347ce1
 	github.com/abhinavxd/youtube-live-chat-downloader/v2 v2.0.3
 	github.com/adeithe/go-twitch v0.3.1
 	github.com/andreykaipov/goobs v1.4.1

go.sum

@@ -615,8 +615,6 @@ github.com/jfreymuth/oggvorbis v1.0.5 h1:u+Ck+R0eLSRhgq8WTmffYnrVtSztJcYrl588DM4
 github.com/jfreymuth/oggvorbis v1.0.5/go.mod h1:1U4pqWmghcoVsCJJ4fRBKv9peUJMBHixthRlBeD6uII=
 github.com/jfreymuth/vorbis v1.0.2 h1:m1xH6+ZI4thH927pgKD8JOH4eaGRm18rEE9/0WKjvNE=
 github.com/jfreymuth/vorbis v1.0.2/go.mod h1:DoftRo4AznKnShRl1GxiTFCseHr4zR9BN3TWXyuzrqQ=
-github.com/joeyak/go-twitch-eventsub/v3 v3.0.0 h1:6BDgmYJynNDyCP7P+wM9jPQnE3leJAi58nohDnzliJ4=
-github.com/joeyak/go-twitch-eventsub/v3 v3.0.0/go.mod h1:rpqOjYP1ftWDj3H4D8fA58AdOpkvK9YvODoduDpPCQU=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
@@ -948,8 +946,6 @@ github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb
 github.com/rymdport/portal v0.4.1 h1:2dnZhjf5uEaeDjeF/yBIeeRo6pNI2QAKm7kq1w/kbnA=
 github.com/rymdport/portal v0.4.1/go.mod h1:kFF4jslnJ8pD5uCi17brj/ODlfIidOxlgUDTO5ncnC4=
 github.com/sagikazarmark/crypt v0.6.0/go.mod h1:U8+INwJo3nBv1m6A/8OBXAq7Jnpspk5AxSgDyEQcea8=
-github.com/scorfly/gokick v1.11.0 h1:4VMpyO8hnPUW4dYgCfOi/YzMgVgj/Rh/OJW7oi2rn+0=
-github.com/scorfly/gokick v1.11.0/go.mod h1:9Tc5+/nwcCLjgDVUeW2K/vhEbkMkN+OE8UQVx4fL1WA=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
@@ -1096,6 +1092,10 @@ github.com/xaionaro-go/fyne/v2 v2.0.0-20250622004601-3a26ee69528a h1:awMQXlaweei
 github.com/xaionaro-go/fyne/v2 v2.0.0-20250622004601-3a26ee69528a/go.mod h1:0GOXKqyvNwk3DLmsFu9v0oYM0ZcD1ysGnlHCerKoAmo=
 github.com/xaionaro-go/go-rtmp v0.0.0-20241009130244-1e3160f27f42 h1:izCjREd+62HDF9FRYqUI7dgJNdUxAIysEuqed8lBcDY=
 github.com/xaionaro-go/go-rtmp v0.0.0-20241009130244-1e3160f27f42/go.mod h1:IuQWd+hy/tLuvuqFX0N9SMZrzOprM8Jvvdu+42RJwk4=
+github.com/xaionaro-go/go-twitch-eventsub/v3 v3.0.0-20250713163657-276e5a5b7adc h1:RwmId5gCeAesuyNKr7BaakXLmHSUxOUqRwV3ME32peU=
+github.com/xaionaro-go/go-twitch-eventsub/v3 v3.0.0-20250713163657-276e5a5b7adc/go.mod h1:rpqOjYP1ftWDj3H4D8fA58AdOpkvK9YvODoduDpPCQU=
+github.com/xaionaro-go/gokick v0.0.0-20250713175656-f538d6396851 h1:P9DQMVL1OWRGKlXZFIwgzivUKsYfQV0y8lfYi5Ek8Yw=
+github.com/xaionaro-go/gokick v0.0.0-20250713175656-f538d6396851/go.mod h1:9Tc5+/nwcCLjgDVUeW2K/vhEbkMkN+OE8UQVx4fL1WA=
 github.com/xaionaro-go/goobs v0.0.0-20241103210141-030e538ac440 h1:hzQ+65oWq54XAqheyJ9E6wt+WH75051w+eLP5zWlD68=
 github.com/xaionaro-go/goobs v0.0.0-20241103210141-030e538ac440/go.mod h1:Fhk0SWHxZ1atNEIm3tUlB93HHRl/v6hWulG/B7BkIOk=
 github.com/xaionaro-go/gorex v0.0.0-20241010205749-bcd59d639c4d h1:9DyH0lboWWzKUwiqGmp9sTZ3bSPhgJHiiWgV+hqY9Uo=

pkg/chatmessagesstorage/sort_and_deduplicate.go

@@ -12,6 +12,9 @@ func msgLess(ctx context.Context, a *api.ChatMessage, b *api.ChatMessage) bool {
 	if a.CreatedAt != b.CreatedAt {
 		return a.CreatedAt.Before(b.CreatedAt)
 	}
+	if a.EventType != b.EventType {
+		return a.EventType < b.EventType
+	}
 	if a.Platform != b.Platform {
 		return a.Platform < b.Platform
 	}

pkg/streamcontrol/event.go

@@ -0,0 +1,34 @@
+package streamcontrol
+
+import "time"
+
+type ChatUserID string
+type ChatMessageID string
+
+type ChatMessage struct {
+	CreatedAt time.Time
+	EventType EventType
+	UserID    ChatUserID
+	Username  string
+	MessageID ChatMessageID
+	Message   string
+	Paid      Money
+}
+
+type EventType int
+
+const (
+	EventTypeUndefined = EventType(iota)
+	EventTypeChatMessage
+	EventTypeCheer
+	EventTypeAutoModHold
+	EventTypeAdBreak
+	EventTypeBan
+	EventTypeFollow
+	EventTypeRaid
+	EventTypeChannelShoutoutReceive
+	EventTypeSubscribe
+	EventTypeStreamOnline
+	EventTypeStreamOffline
+	EventTypeOther
+)

pkg/streamcontrol/kick/chat_handler_obsolete.go

@@ -148,6 +148,7 @@ func (h *ChatHandlerOBSOLETE) sendMessage(
 	select {
 	case h.messagesOutChan <- streamcontrol.ChatMessage{
 		CreatedAt: msg.CreatedAt,
+		EventType: streamcontrol.EventTypeChatMessage,
 		UserID:    streamcontrol.ChatUserID(fmt.Sprintf("%d", msg.UserID)),
 		Username:  msg.Sender.Slug,
 		MessageID: streamcontrol.ChatMessageID(msg.ID),

pkg/streamcontrol/kick/kick.go

@@ -78,17 +78,16 @@ func New(
 
 func (k *Kick) initChatHandler(
 	ctx context.Context,
-) {
+) error {
 	if !k.ChatHandlerLocker.Lock(ctx) {
-		logger.Debugf(ctx, "initChatHandler: cancelled (case #0)")
-		return
+		return ctx.Err()
 	}
 
 	chatHandler, err := k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel, k.onChatHandlerClose)
 	if err == nil {
 		k.ChatHandlerLocker.Unlock()
 		k.ChatHandler = chatHandler
-		return
+		return nil
 	}
 
 	go func() {
@@ -110,6 +109,7 @@ func (k *Kick) initChatHandler(
 		}
 		k.ChatHandler = chatHandler
 	}()
+	return nil
 }
 
 func (k *Kick) onChatHandlerClose(
@@ -194,18 +194,21 @@ func (k *Kick) getAccessTokenNoLock(
 	codeVerifierSHA256 := sha256.Sum256([]byte(codeVerifier))
 	codeChallenge := base64.URLEncoding.EncodeToString(codeVerifierSHA256[:])
 
+	scopes := []gokick.Scope{
+		gokick.ScopeUserRead,
+		gokick.ScopeChannelRead,
+		gokick.ScopeChannelWrite,
+		gokick.ScopeChatWrite,
+		gokick.ScopeStremkeyRead,
+		gokick.ScopeEventSubscribe,
+		gokick.ScopeModerationBan,
+	}
+	logger.Debugf(ctx, "scopes: %v", scopes)
 	authURL, err := k.getClient().GetAuthorize(
 		redirectURL,
 		"EMPTY",
 		codeChallenge,
-		[]gokick.Scope{
-			gokick.ScopeUserRead,
-			gokick.ScopeChannelRead,
-			gokick.ScopeChannelWrite,
-			gokick.ScopeChatWrite,
-			gokick.ScopeStremkeyRead,
-			gokick.ScopeEventSubscribe,
-		},
+		scopes,
 	)
 	if err != nil {
 		return fmt.Errorf("unable to get an authorization endpoint URL: %w", err)
@@ -574,25 +577,31 @@ func (k *Kick) prepare(ctx context.Context) (_err error) {
 func (k *Kick) prepareNoLock(ctx context.Context) error {
 	err := k.getAccessTokenIfNeeded(ctx)
 	if err != nil {
-		return err
+		return fmt.Errorf("getAccessTokenIfNeeded: %w", err)
 	}
 
 	k.lazyInitOnce.Do(func() {
-		k.initChannelInfo(ctx)
-		k.initChatHandler(ctx)
+		if err = k.initChannelInfo(ctx); err != nil {
+			err = fmt.Errorf("initChannelInfo: %w", err)
+			return
+		}
+		if err = k.initChatHandler(ctx); err != nil {
+			err = fmt.Errorf("initChatHandler: %w", err)
+			return
+		}
 	})
 	return err
 }
 
 func (k *Kick) initChannelInfo(
 	ctx context.Context,
-) {
+) error {
 	var channel *gokick.ChannelResponse
 	cache := CacheFromCtx(ctx)
 	if chanInfo := cache.GetChanInfo(); chanInfo != nil && chanInfo.Slug == k.CurrentConfig.Config.Channel {
 		logger.Debugf(ctx, "reuse the cache, instead of querying channel info")
 		k.Channel = chanInfo
-		return
+		return nil
 	}
 
 	for {
@@ -613,7 +622,7 @@ func (k *Kick) initChannelInfo(
 			cache.SetChanInfo(channel)
 		}
 		k.Channel = channel
-		return
+		return nil
 	}
 }
 

pkg/streamcontrol/stream_control.go

@@ -106,22 +106,11 @@ type StreamStatus struct {
 	CustomData   any        `json:",omitempty"`
 }
 
-type ChatUserID string
-type ChatMessageID string
-
-type ChatMessage struct {
-	CreatedAt time.Time
-	UserID    ChatUserID
-	Username  string
-	MessageID ChatMessageID
-	Message   string
-	Paid      Money
-}
-
 type Currency int
 
 const (
 	CurrencyNone = Currency(iota)
+	CurrencyBits
 	CurrencyUSD
 	CurrencyOther
 )

pkg/streamcontrol/twitch/auth/client_code.go

@@ -3,6 +3,9 @@ package auth
 import (
 	"context"
 	"fmt"
+	"maps"
+	"slices"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -17,6 +20,99 @@ import (
 
 type OAuthHandler func(context.Context, oauthhandler.OAuthHandlerArgument) error
 
+func getScopes() []string {
+	scopes := map[string]struct{}{
+		"chat:read": {},
+		"chat:edit": {},
+
+		"analytics:read:extensions": {},
+		"analytics:read:games":      {},
+
+		"bits:read": {},
+
+		"channel:bot":                {},
+		"channel:manage:ads":         {},
+		"channel:read:ads":           {},
+		"channel:manage:broadcast":   {},
+		"channel:read:charity":       {},
+		"channel:edit:commercial":    {},
+		"channel:read:editors":       {},
+		"channel:manage:extensions":  {},
+		"channel:read:goals":         {},
+		"channel:read:guest_star":    {},
+		"channel:manage:guest_star":  {},
+		"channel:read:hype_train":    {},
+		"channel:manage:moderators":  {},
+		"channel:read:polls":         {},
+		"channel:manage:polls":       {},
+		"channel:read:predictions":   {},
+		"channel:manage:predictions": {},
+		"channel:manage:raids":       {},
+		"channel:read:redemptions":   {},
+		"channel:manage:redemptions": {},
+		"channel:manage:schedule":    {},
+		"channel:read:stream_key":    {},
+		"channel:read:subscriptions": {},
+		"channel:manage:videos":      {},
+		"channel:read:vips":          {},
+		"channel:manage:vips":        {},
+		"channel:moderate":           {},
+
+		"clips:edit": {},
+
+		"moderation:read": {},
+
+		"moderator:manage:announcements":    {},
+		"moderator:manage:automod":          {},
+		"moderator:read:automod_settings":   {},
+		"moderator:manage:automod_settings": {},
+		"moderator:read:banned_users":       {},
+		"moderator:manage:banned_users":     {},
+		"moderator:read:blocked_terms":      {},
+		"moderator:read:chat_messages":      {},
+		"moderator:manage:blocked_terms":    {},
+		"moderator:manage:chat_messages":    {},
+		"moderator:read:chat_settings":      {},
+		"moderator:manage:chat_settings":    {},
+		"moderator:read:chatters":           {},
+		"moderator:read:followers":          {},
+		"moderator:read:guest_star":         {},
+		"moderator:manage:guest_star":       {},
+		"moderator:read:moderators":         {},
+		"moderator:read:shield_mode":        {},
+		"moderator:manage:shield_mode":      {},
+		"moderator:read:shoutouts":          {},
+		"moderator:manage:shoutouts":        {},
+		"moderator:read:suspicious_users":   {},
+		"moderator:read:unban_requests":     {},
+		"moderator:manage:unban_requests":   {},
+		"moderator:read:vips":               {},
+		"moderator:read:warnings":           {},
+		"moderator:manage:warnings":         {},
+
+		"user:bot":                     {},
+		"user:edit":                    {},
+		"user:edit:broadcast":          {},
+		"user:read:blocked_users":      {},
+		"user:manage:blocked_users":    {},
+		"user:read:broadcast":          {},
+		"user:read:chat":               {},
+		"user:manage:chat_color":       {},
+		"user:read:email":              {},
+		"user:read:emotes":             {},
+		"user:read:follows":            {},
+		"user:read:moderated_channels": {},
+		"user:read:subscriptions":      {},
+		"user:read:whispers":           {},
+		"user:manage:whispers":         {},
+		"user:write:chat":              {},
+	}
+
+	scopesStrings := slices.Collect(maps.Keys(scopes))
+	sort.Strings(scopesStrings)
+	return scopesStrings
+}
+
 func NewClientCode(
 	ctx context.Context,
 	clientID string,
@@ -69,14 +165,7 @@ func NewClientCode(
 				authURL := GetAuthorizationURL(
 					&helix.AuthorizationURLParams{
 						ResponseType: "code", // or "token"
-						Scopes: []string{
-							"user:read:chat",
-							"chat:read",
-							"chat:edit",
-							"channel:manage:broadcast",
-							"moderator:manage:chat_messages",
-							"moderator:manage:banned_users",
-						},
+						Scopes:       getScopes(),
 					},
 					clientID,
 					RedirectURI(listenPort),

pkg/streamcontrol/twitch/chat_handler_irc.go

@@ -93,6 +93,7 @@ func newChatHandlerIRC(
 				select {
 				case h.messagesOutChan <- streamcontrol.ChatMessage{
 					CreatedAt: ev.CreatedAt,
+					EventType: streamcontrol.EventTypeChatMessage,
 					UserID:    streamcontrol.ChatUserID(ev.Sender.Username),
 					Username:  ev.Sender.Username,
 					MessageID: streamcontrol.ChatMessageID(ev.ID),