package twitch import ( "bytes" "context" "crypto/rand" "embed" "encoding/json" "errors" "fmt" "io" "log" "net/http" "net/url" "os" "os/signal" "path" "strings" "time" "codeberg.org/arimelody/ari-stream-tools/broadcast" "codeberg.org/arimelody/ari-stream-tools/config" "codeberg.org/arimelody/ari-stream-tools/twitch/api" "github.com/gin-gonic/gin" "github.com/gorilla/websocket" "golang.org/x/oauth2" "golang.org/x/oauth2/twitch" ) type ( twitchLabel struct { Text string C chan string Broadcast broadcast.BroadcastChannel[string] } twitchLabels struct { LatestFollower *twitchLabel LatestSubscriber *twitchLabel LatestCheer *twitchLabel } ServiceOptions struct { Port int16 } serviceConfig struct { ChannelName string `json:"channel_name"` ClientID string `json:"client_id"` ClientSecret string `json:"client_secret"` } Service struct { port int16 labels *twitchLabels channelName string channelID string clientID string clientSecret string oauthConfig *oauth2.Config oauthState string oauthToken *oauth2.Token eventSubSession *api.EventSubSession } ) const ( LABEL_LATEST_FOLLOWER string = "latest-follower" LABEL_LATEST_SUBSCRIBER string = "latest-subscriber" LABEL_LATEST_CHEER string = "latest-cheer" ) var DATA_PATH string = path.Join(config.CONFIG_DIR, "twitch") var CONFIG_FILEPATH string = path.Join(DATA_PATH, "twitch-config.json") var AUTH_FILEPATH string = path.Join(DATA_PATH, "twitch-auth") //go:embed public var publicFS embed.FS //go:embed pages var pagesFS embed.FS func New(ctx context.Context, opts ServiceOptions) (*Service, error) { if err := os.MkdirAll(DATA_PATH, 0750); err != nil { panic(err) } if err := os.MkdirAll(path.Join(DATA_PATH, "state"), 0750); err != nil { panic(err) } config := serviceConfig{} if configFile, err := os.OpenFile(CONFIG_FILEPATH, os.O_CREATE | os.O_RDWR, 0600); err != nil { return nil, fmt.Errorf("open %s: %v", CONFIG_FILEPATH, err) } else { defer configFile.Close() stat, err := configFile.Stat() if err != nil { return nil, fmt.Errorf("stat %s: %v", CONFIG_FILEPATH, err) } if stat.Size() == 0 { enc := json.NewEncoder(configFile) enc.SetIndent("", "\t") if err := enc.Encode(&config); err != nil { return nil, fmt.Errorf("write %s: %v", CONFIG_FILEPATH, err) } return nil, fmt.Errorf("Config file is empty: %s", CONFIG_FILEPATH) } else if err := json.NewDecoder(configFile).Decode(&config); err != nil { return nil, fmt.Errorf("read %s: %v", CONFIG_FILEPATH, err) } } if len(config.ChannelName) == 0 { return nil, errors.New("config: channel_name cannot be empty") } if len(config.ClientID) == 0 { return nil, errors.New("config: client_id cannot be empty") } if len(config.ClientSecret) == 0 { return nil, errors.New("config: client_secret cannot be empty") } latestFollowerC := make(chan string) latestFollowerBroadcast := broadcast.NewBroadcastChannel( ctx, latestFollowerC) latestSubscriberC := make(chan string) latestSubscriberBroadcast := broadcast.NewBroadcastChannel( ctx, latestSubscriberC) latestCheerC := make(chan string) latestCheerBroadcast := broadcast.NewBroadcastChannel( ctx, latestCheerC) srv := &Service{ port: opts.Port, labels: &twitchLabels{ LatestFollower: &twitchLabel{ Text: "some_follower", C: latestFollowerC, Broadcast: latestFollowerBroadcast, }, LatestSubscriber: &twitchLabel{ Text: "some_subscriber", C: latestSubscriberC, Broadcast: latestSubscriberBroadcast, }, LatestCheer: &twitchLabel{ Text: "some_cheer", C: latestCheerC, Broadcast: latestCheerBroadcast, }, }, channelName: config.ChannelName, clientID: config.ClientID, clientSecret: config.ClientSecret, oauthConfig: &oauth2.Config{ ClientID: config.ClientID, ClientSecret: config.ClientSecret, Endpoint: twitch.Endpoint, Scopes: []string{ "moderator:read:followers", "user:read:chat", "user:bot", "channel:bot", "channel:read:subscriptions", "bits:read", "channel:read:redemptions", "channel:read:polls", "channel:read:predictions", "channel:read:hype_train", "moderator:read:shoutouts", }, RedirectURL: fmt.Sprintf("http://localhost:%d/twitch/auth", opts.Port), }, } if authFile, err := os.OpenFile(AUTH_FILEPATH, os.O_RDONLY, 0600); err != nil { if !os.IsNotExist(err) { log.Fatalf("open %s: %v", AUTH_FILEPATH, err) } } else { defer authFile.Close() srv.oauthToken = &oauth2.Token{} err = json.NewDecoder(authFile).Decode(srv.oauthToken) if err != nil { log.Printf("read %s: %v", AUTH_FILEPATH, err) } } if data, err := os.ReadFile(path.Join(DATA_PATH, "state", LABEL_LATEST_FOLLOWER)); err == nil { srv.labels.LatestFollower.Text = string(data) log.Printf("Loaded latest follower: %s", string(data)) } if data, err := os.ReadFile(path.Join(DATA_PATH, "state", LABEL_LATEST_SUBSCRIBER)); err == nil { srv.labels.LatestSubscriber.Text = string(data) log.Printf("Loaded latest subscriber: %s", string(data)) } if data, err := os.ReadFile(path.Join(DATA_PATH, "state", LABEL_LATEST_CHEER)); err == nil { srv.labels.LatestCheer.Text = string(data) log.Printf("Loaded latest cheer: %s", string(data)) } return srv, nil } func (srv *Service) BindRoutes(group *gin.RouterGroup) { group.GET("/public/*path", func(ctx *gin.Context) { path := strings.TrimPrefix(ctx.Request.URL.Path, "/twitch/") http.ServeFileFS(ctx.Writer, ctx.Request, publicFS, path) }) group.GET("/login", func(ctx *gin.Context) { srv.oauthState = rand.Text() authCodeURL := srv.oauthConfig.AuthCodeURL(srv.oauthState) ctx.Redirect(http.StatusTemporaryRedirect, authCodeURL) }) group.GET("/auth", func(ctx *gin.Context) { code := ctx.Query("code") scope := ctx.Query("scope") resState := ctx.Query("state") if len(code) == 0 { ctx.String(http.StatusBadRequest, "code cannot be empty"); return } if len(scope) == 0 { ctx.String(http.StatusBadRequest, "scope cannot be empty"); return } if resState != srv.oauthState { ctx.String(http.StatusBadRequest, "state mismatch"); return } token, err := srv.oauthConfig.Exchange(ctx, code) if err != nil { log.Printf("Could not exchange OAuth2 code: %v", err) ctx.String(http.StatusBadRequest, "Could not exchange OAuth2 code.") return } srv.oauthToken = token authFile, err := os.OpenFile(AUTH_FILEPATH, os.O_CREATE | os.O_RDWR, 0600) if err != nil { log.Printf("open %s: %v", AUTH_FILEPATH, err) ctx.String(http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) return } defer authFile.Close() authFile.Truncate(0) json.NewEncoder(authFile).Encode(srv.oauthToken) ctx.String( http.StatusOK, "Authentication successful! You may now close this tab.", ) go srv.start(ctx) }) group.GET("/sse", func(ctx *gin.Context) { ctx.Header("connection", "keep-alive") listeningTo := ctx.Query("l") var label *twitchLabel switch listeningTo { case LABEL_LATEST_FOLLOWER: label = srv.labels.LatestFollower case LABEL_LATEST_SUBSCRIBER: label = srv.labels.LatestSubscriber case LABEL_LATEST_CHEER: label = srv.labels.LatestCheer default: ctx.String(http.StatusBadRequest, "Unknown label %s", listeningTo) return } labelUpdate := label.Broadcast.Subscribe() defer label.Broadcast.Cancel(labelUpdate) ctx.SSEvent("update", label.Text) ticker := time.NewTicker(10 * time.Millisecond) ctx.Stream(func(w io.Writer) bool { select { case text := <-labelUpdate: ctx.SSEvent("update", text) case <-ticker.C: } return true }) }) group.GET("/label", func(ctx *gin.Context) { http.ServeFileFS(ctx.Writer, ctx.Request, pagesFS, "pages/labels.html") }) } func (srv *Service) Run(ctx context.Context) { if srv.oauthToken == nil || !srv.oauthToken.Valid() { log.Printf("Log in with Twitch: http://localhost:%d/twitch/login", srv.port) } else { srv.start(ctx) } } func (srv *Service) start(ctx context.Context) { userIDs, err := srv.userIDsFromNames([]string{ srv.channelName }) if err != nil { log.Printf( "Failed to resolve Twitch username \"%s\": %v", srv.channelName, err, ) return } if len(userIDs) == 0 { fmt.Printf("Failed to resolve username \"%s\", it may not exist?", srv.channelName) return } srv.channelID = userIDs[0] interrupt := make(chan os.Signal, 1) signal.Notify(interrupt, os.Interrupt) u, err := url.Parse(api.EVENTSUB_URL + "?keepalive_timeout_seconds=600") if err != nil { panic(err) } c, _, err := websocket.DefaultDialer.Dial(u.String(), http.Header{ "Authorization": []string{ "Bearer " + srv.clientSecret }, }) if err != nil { log.Printf("Failed to connect to Twitch: %v", err) } defer c.Close() failed := make(chan error) go func() { for { _, rawData, err := c.ReadMessage() if err != nil { failed <- err; return } var message api.EventSubMessage err = json.Unmarshal(rawData, &message) if err != nil { failed <- fmt.Errorf("parse JSON: %v", err) return } if message.Payload.Session != nil { if err := srv.registerEventSubSession(message.Payload.Session); err != nil { failed <- fmt.Errorf("register eventsub session: %v", err) return } srv.subscribeToDefaultEvents(ctx) log.Printf("Connected to Twitch as %s.", srv.channelName) } if message.Metadata.MessageType == string(api.NOTIFICATION) { if message.Payload.Subscription == nil { continue } if err := srv.handleNotification(&message.Payload); err != nil { log.Printf("Failed to handle %s event: %v", message.Payload.Subscription.Type, err) } continue } } }() select { case err := <-failed: log.Fatalf("Twitch error: %v", err) case <-ctx.Done(): } } func (srv *Service) registerEventSubSession(session *api.EventSubSession) error { srv.eventSubSession = session return nil } func (srv *Service) handleNotification(payload *api.EventSubPayload) error { var err error jsonData, err := json.Marshal(payload.Event) if err != nil { return err } switch payload.Subscription.Type { case "channel.follow": var event api.FollowEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.FollowEvent: %v", err) } log.Printf("New follow: %s", event.UserLogin) srv.labels.LatestFollower.Text = event.UserLogin srv.labels.LatestFollower.C <- event.UserLogin if err := os.WriteFile( path.Join(DATA_PATH, "state", LABEL_LATEST_FOLLOWER), []byte(event.UserLogin), 0640, ); err != nil { return fmt.Errorf("Failed to write %s state: %v", LABEL_LATEST_FOLLOWER, err) } case "channel.subscribe": var event api.SubscribeEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.SubscribeEvent: %v", err) } log.Printf("New subscription: %s", event.UserLogin) srv.labels.LatestSubscriber.Text = event.UserLogin srv.labels.LatestSubscriber.C <- event.UserLogin if err := os.WriteFile( path.Join(DATA_PATH, "state", LABEL_LATEST_SUBSCRIBER), []byte(event.UserLogin), 0640, ); err != nil { return fmt.Errorf("Failed to write %s state: %v", LABEL_LATEST_SUBSCRIBER, err) } case "channel.cheer": var event api.CheerEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.CheerEvent: %v", err) } log.Printf("%s cheered x%d bits: %s", event.UserLogin, event.Bits, event.Message) srv.labels.LatestCheer.Text = event.UserLogin srv.labels.LatestCheer.C <- event.UserLogin if err := os.WriteFile( path.Join(DATA_PATH, "state", LABEL_LATEST_CHEER), []byte(event.UserLogin), 0640, ); err != nil { return fmt.Errorf("Failed to write %s state: %v", LABEL_LATEST_CHEER, err) } case "channel.raid": var event api.RaidEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.RaidEvent: %v", err) } log.Printf("%s is now raiding with %d viewers!", event.FromUserLogin, event.Viewers) case "channel.channel_points_custom_reward_redemption.add": var event api.ChannelPointCustomRewardRedeemEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.ChannelPointCustomRewardRedeemEvent: %v", err) } log.Printf( "%s just redeemed %s for %d channel points.", event.UserLogin, event.Reward.Title, event.Reward.Cost, ) case "channel.shoutout.create": var event api.ShoutoutCreate err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.ShoutoutCreate: %v", err) } log.Printf( "%s gave a shoutout to %s.", event.FromUserLogin, event.ToUserLogin, ) case "channel.chat.message": var event api.ChatEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.ChatEvent: %v", err) } if event.Cheer != nil { return nil } log.Printf( "[%s] %s: %s", event.MessageID, event.ChatterLogin, event.Message.Text, ) case "channel.chat.message_delete": var event api.ChatDeleteEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.ChatDeleteEvent: %v", err) } log.Printf( "Message %s by %s deleted.", event.MessageID, event.TargetLogin, ) } return nil } func (srv *Service) subscribeToEvent( ctx context.Context, subscriptionType string, version string, condition map[string]string, sessionID string, ) error { type ( Transport struct { Method string `json:"method"` SessionID string `json:"session_id"` } Request struct { Type string `json:"type"` Version string `json:"version"` Condition map[string]string `json:"condition"` Transport Transport `json:"transport"` } ResponseData struct { ID string `json:"id"` Status string `json:"status"` Type string `json:"type"` Version string `json:"version"` Condition map[string]string `json:"condition"` CreatedAt string `json:"created_at"` Transport Transport `json:"transport"` Cost int `json:"cost"` } Response struct { Data []ResponseData `json:"data"` Total int `json:"total"` TotalCost int `json:"total_cost"` MaxTotalCost int `json:"max_total_cost"` } ) bodyBytes, err := json.Marshal(Request{ Type: subscriptionType, Version: version, Condition: condition, Transport: Transport{ Method: "websocket", SessionID: sessionID, }, }) body := bytes.NewBuffer(bodyBytes) client := srv.oauthConfig.Client(ctx, srv.oauthToken) req, err := http.NewRequest( "POST", api.BASE_URL + "/eventsub/subscriptions", body, ) if err != nil { return err } req.Header.Set("Content-Type", "application/json") req.Header.Set("Client-Id", srv.clientID) req.Header.Set("Authorization", "Bearer " + srv.oauthToken.AccessToken) res, err := client.Do(req) if err != nil { return err } if res.StatusCode != http.StatusAccepted { body, _ := io.ReadAll(res.Body) return fmt.Errorf("%s: %s", res.Status, string(body)) } return nil } func (srv *Service) subscribeToDefaultEvents(ctx context.Context) { // channel.follow if err := srv.subscribeToEvent( ctx, "channel.follow", "2", map[string]string{ "broadcaster_user_id": srv.channelID, "moderator_user_id": srv.channelID, }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.follow: %v", err) } // channel.subscribe if err := srv.subscribeToEvent( ctx, "channel.subscribe", "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.subscribe: %v", err) } // channel.cheer if err := srv.subscribeToEvent( ctx, "channel.cheer", "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.cheer: %v", err) } // channel.raid if err := srv.subscribeToEvent( ctx, "channel.raid", "1", map[string]string{ "to_broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.raid: %v", err) } // channel.channel_points_custom_reward_redemption.add if err := srv.subscribeToEvent( ctx, "channel.channel_points_custom_reward_redemption.add", "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.channel_points_custom_reward_redemption.add: %v", err) } // channel.shoutout.create if err := srv.subscribeToEvent( ctx, "channel.shoutout.create", "1", map[string]string{ "broadcaster_user_id": srv.channelID, "moderator_user_id": srv.channelID, }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.shoutout.create: %v", err) } // channel.chat.message if err := srv.subscribeToEvent( ctx, "channel.chat.message", "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.chat.message: %v", err) } // channel.chat.message_delete if err := srv.subscribeToEvent( ctx, "channel.chat.message_delete", "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ); err != nil { log.Printf("Failed to subscribe to channel.chat.message_delete: %v", err) } } func (srv *Service) userIDsFromNames(usernames []string) ([]string, error) { if len(usernames) == 0 { return []string{}, nil } url := strings.Builder{} url.WriteString(api.BASE_URL) url.WriteString("/users?login=") url.WriteString(usernames[0]) for _, username := range usernames { url.WriteString("&login=") url.WriteString(username) } req, err := http.NewRequest("GET", url.String(), nil) if err != nil { return nil, err } req.Header.Set("Client-Id", srv.clientID) req.Header.Set("Authorization", "Bearer " + srv.oauthToken.AccessToken) res, err := http.DefaultClient.Do(req) if err != nil { return nil, err } if res.StatusCode != http.StatusOK { body, _ := io.ReadAll(res.Body) return nil, fmt.Errorf("%s: %s", res.Status, string(body)) } type ( UserData struct { ID string `json:"id"` Login string `json:"login"` DisplayName string `json:"display_name"` Type string `json:"type"` BroadcasterType string `json:"broadcaster_type"` Description string `json:"description"` ProfileImageURL string `json:"profile_image_url"` OfflineImageURL string `json:"offline_image_url"` ViewCount int `json:"view_count"` Email string `json:"email"` CreatedAt string `json:"created_at"` } Response struct { Users []UserData `json:"data"` } ) data := Response{} if err := json.NewDecoder(res.Body).Decode(&data); err != nil { return nil, err } userIDs := []string{} for _, user := range data.Users { userIDs = append(userIDs, user.ID) } return userIDs, nil }