package twitch import ( "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 channelName string channelID string clientID string clientSecret string oauthConfig *oauth2.Config oauthState string oauthToken *oauth2.Token eventSubSession *api.EventSubSession labels *twitchLabels } ) 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] srv.startWebsocketListener(ctx) } func (srv *Service) startWebsocketListener(ctx context.Context) { 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.Printf("Twitch error: %v", err) case <-ctx.Done(): } }