add twitch API stuff, add follow/sub/cheer labels

This commit is contained in:
ari melody 2026-07-24 03:54:29 +01:00
parent 8a7210ad56
commit 626540e728
Signed by: ari
GPG key ID: 60B5F0386E3DDB7E
17 changed files with 1143 additions and 3 deletions

657
twitch/twitch.go Normal file
View file

@ -0,0 +1,657 @@
package twitch
import (
"bytes"
"context"
"crypto/rand"
"embed"
"encoding/json"
"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"
)
type (
twitchLabel struct {
Text string
C chan string
Broadcast broadcast.BroadcastChannel[string]
}
twitchLabels struct {
LatestFollower *twitchLabel
LatestSubscriber *twitchLabel
LatestCheer *twitchLabel
}
ServiceOptions struct {
Port int16
ChannelName string
ClientID string
ClientSecret string
}
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 (
TWITCH_EVENTSUB_URL string = "wss://eventsub.wss.twitch.tv/ws"
TWITCH_API_BASE string = "https://api.twitch.tv/helix"
TWITCH_OAUTH2_URL string = "https://id.twitch.tv/oauth2/token"
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")
//go:embed public
var publicFS embed.FS
//go:embed pages
var pagesFS embed.FS
func New(ctx context.Context, opts ServiceOptions) (*Service, error) {
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{
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: opts.ChannelName,
clientID: opts.ClientID,
clientSecret: opts.ClientSecret,
oauthConfig: &oauth2.Config{
ClientID: opts.ClientID,
ClientSecret: opts.ClientSecret,
Endpoint: oauth2.Endpoint{
AuthURL: "https://id.twitch.tv/oauth2/authorize",
TokenURL: "https://id.twitch.tv/oauth2/token",
},
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 err := os.MkdirAll(DATA_PATH, 0750); err != nil { panic(err) }
if err := os.MkdirAll(path.Join(DATA_PATH, "state"), 0750); err != nil { panic(err) }
authFile, err := os.OpenFile(path.Join(DATA_PATH, "twitch-auth.json"), os.O_RDONLY, 0600)
if err != nil {
if !os.IsNotExist(err) {
log.Fatalf("Failed to open twitch-auth.json: %v", err)
}
} else {
srv.oauthToken = &oauth2.Token{}
err = json.NewDecoder(authFile).Decode(srv.oauthToken)
if err != nil {
log.Printf("Failed to read twitch-auth.json: %v", err)
}
}
authFile.Close()
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("/auth", func(ctx *gin.Context) {
code := ctx.Query("code")
scope := ctx.Query("scope")
resState := ctx.Query("state")
if len(code) == 0 || len(scope) == 0 || resState != srv.oauthState {
ctx.String(http.StatusBadRequest, http.StatusText(http.StatusBadRequest))
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(path.Join(DATA_PATH, "twitch-auth.json"), os.O_CREATE | os.O_RDWR, 0600)
if err != nil {
log.Printf("Failed to open twitch-auth.json: %v", 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("/labels", 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.oauthState = rand.Text()
authCodeURL := srv.oauthConfig.AuthCodeURL(srv.oauthState)
log.Printf("Log in with Twitch: %s", authCodeURL)
} 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]
log.Printf("Connecting to Twitch as %s...", srv.channelName)
interrupt := make(chan os.Signal, 1)
signal.Notify(interrupt, os.Interrupt)
u, err := url.Parse(TWITCH_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()
}
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(
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 := http.DefaultClient
req, err := http.NewRequest(
"POST",
TWITCH_API_BASE + "/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() {
// channel.follow
if err := srv.subscribeToEvent(
"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(
"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(
"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(
"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(
"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(
"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(
"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(
"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(TWITCH_API_BASE)
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
}