diff --git a/.air.toml b/.air.toml index 23141a3..ceab2de 100644 --- a/.air.toml +++ b/.air.toml @@ -7,10 +7,10 @@ tmp_dir = "tmp" [build] args_bin = [] - bin = "./tmp/main" - cmd = "go build -o ./tmp/main ." + bin = "./tmp/stream-tools" + cmd = "go build -o ./tmp/stream-tools ./cmd/stream-tools" delay = 1000 - entrypoint = ["./tmp/main"] + entrypoint = ["./tmp/stream-tools"] exclude_dir = ["assets", "tmp", "vendor", "testdata"] exclude_file = [] exclude_regex = ["_test.go"] diff --git a/cmd/satellite/main.go b/cmd/satellite/main.go new file mode 100644 index 0000000..df4ea4b --- /dev/null +++ b/cmd/satellite/main.go @@ -0,0 +1,5 @@ +package main + +func main() { + // TODO: build satellite service for offline-fetching subscribers and cheers +} diff --git a/main.go b/cmd/stream-tools/main.go similarity index 88% rename from main.go rename to cmd/stream-tools/main.go index 57332e9..64791cc 100644 --- a/main.go +++ b/cmd/stream-tools/main.go @@ -43,10 +43,14 @@ func main() { srv := gin.New() srv.SetTrustedProxies([]string{"127.0.0.1", "::1"}) - learningService := learning.New(ctx, learning.ServiceConfig{ + learningService, err := learning.New(ctx, learning.ServiceConfig{ TitleFilePath: "learning-title.txt", }) - learningService.BindRoutes(srv.Group("/learning")) + if err != nil { + log.Printf("Failed to create learning service: %v", err) + } else { + learningService.BindRoutes(srv.Group("/learning")) + } musicService, err := music.New(ctx) if err != nil { diff --git a/learning/learning.go b/learning/learning.go index fb8a56f..6f8247c 100644 --- a/learning/learning.go +++ b/learning/learning.go @@ -3,6 +3,7 @@ package learning import ( "context" "embed" + "fmt" "io" "log" "net/http" @@ -36,7 +37,7 @@ type ( } ) -func New(ctx context.Context, cfg ServiceConfig) *Service { +func New(ctx context.Context, cfg ServiceConfig) (*Service, error) { titleText := []byte{} titleUpdated := make(chan string) @@ -45,10 +46,10 @@ func New(ctx context.Context, cfg ServiceConfig) *Service { os.WriteFile(cfg.TitleFilePath, []byte("Untitled"), 0644) titleText = []byte("Untitled") } else if err != nil { - log.Fatalf("Failed to stat title file: %v", err) + return nil, fmt.Errorf("stat %s: %v", cfg.TitleFilePath, err) } else { titleText, err = os.ReadFile(cfg.TitleFilePath) - if err != nil { log.Fatalf("Failed to read title file: %v", err) } + if err != nil { return nil, fmt.Errorf("read %s: %v", cfg.TitleFilePath, err) } } srv := &Service{ @@ -60,7 +61,7 @@ func New(ctx context.Context, cfg ServiceConfig) *Service { titleUpdatedBroadcast: broadcast.NewBroadcastChannel(ctx, titleUpdated), } - return srv + return srv, nil } func (srv *Service) BindRoutes(group *gin.RouterGroup) { diff --git a/learning/public/cards.js b/learning/public/cards.js index a25da1a..eee14ed 100644 --- a/learning/public/cards.js +++ b/learning/public/cards.js @@ -43,8 +43,6 @@ async function updateTitle() { console.error("Connection lost, or an error has occurred."); console.log("Attempting to reconnect..."); - // TODO: stop reconnecting after 10 failed attempts - setTimeout(() => { updateTitle(); }, 1000); diff --git a/music/music.go b/music/music.go index bab21b8..d3d2506 100644 --- a/music/music.go +++ b/music/music.go @@ -225,7 +225,7 @@ func (srv *Service) Run(ctx context.Context) { select { case err := <-failed: - log.Fatalf("MPRIS/D-Bus error: %v", err) + log.Printf("MPRIS/D-Bus error: %v", err) case <-ctx.Done(): } } diff --git a/music/public/music.js b/music/public/music.js index 115452b..dfd2e42 100644 --- a/music/public/music.js +++ b/music/public/music.js @@ -37,8 +37,6 @@ async function sync() { console.error("Connection lost, or an error has occurred."); console.log("Attempting to reconnect..."); - // TODO: stop reconnecting after 10 failed attempts - setTimeout(() => { sync(); }, 1000); diff --git a/twitch/api.go b/twitch/api.go new file mode 100644 index 0000000..f9d265b --- /dev/null +++ b/twitch/api.go @@ -0,0 +1,444 @@ +package twitch + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "net/url" + "os" + "path" + "strconv" + + "codeberg.org/arimelody/ari-stream-tools/twitch/api" +) + +type ( + FullUser 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"` + } + + GetUsersResponse struct { + Data []FullUser `json:"data"` + } +) +func (srv *Service) getUsers(ids []string, logins []string) (*GetUsersResponse, error) { + if len(ids) == 0 && len(logins) == 0 { return nil, nil } + + url, err := url.Parse(api.BASE_URL + "/users?" + url.Values{ + "id": ids, "login": logins, + }.Encode()) + + 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)) + } + + data := &GetUsersResponse{} + if err := json.NewDecoder(res.Body).Decode(data); err != nil { + return nil, err + } + + return data, nil +} + +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.UserName) + srv.labels.LatestFollower.Text = event.UserName + srv.labels.LatestFollower.C <- event.UserName + + if err := os.WriteFile( + path.Join(DATA_PATH, "state", LABEL_LATEST_FOLLOWER), + []byte(event.UserName), 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.UserName) + srv.labels.LatestSubscriber.Text = event.UserName + srv.labels.LatestSubscriber.C <- event.UserName + + if err := os.WriteFile( + path.Join(DATA_PATH, "state", LABEL_LATEST_SUBSCRIBER), + []byte(event.UserName), 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.UserName, event.Bits, event.Message) + srv.labels.LatestCheer.Text = event.UserName + srv.labels.LatestCheer.C <- event.UserName + + if err := os.WriteFile( + path.Join(DATA_PATH, "state", LABEL_LATEST_CHEER), + []byte(event.UserName), 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.FromUserName, 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.UserName, + 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.FromUserName, + event.ToUserName, + ) + + 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) + } +} + +type ( + Follower struct { + FollowedAt string `json:"followed_at"` + UserID string `json:"user_id"` + UserLogin string `json:"user_login"` + UserName string `json:"user_name"` + } + + Pagination struct { + Cursor string `json:"cursor"` + } + + GetFollowersResponse struct { + Data []Follower `json:"data"` + Pagination Pagination `json:"pagination"` + Total int `json:"total"` + Points int `json:"points"` + } +) +func (srv *Service) getFollowers( + ctx context.Context, + broadcasterID string, + limit int, +) (*GetFollowersResponse, error) { + client := srv.oauthConfig.Client(ctx, srv.oauthToken) + u, err := url.Parse(api.BASE_URL + "/channels/followers?" + url.Values{ + "broadcaster_id": []string{ broadcasterID }, + "first": []string { strconv.Itoa(limit) }, + }.Encode()) + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { return nil, 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 nil, err } + + if res.StatusCode != http.StatusOK { + body, _ := io.ReadAll(res.Body) + return nil, fmt.Errorf("%s: %s", res.Status, string(body)) + } + + resData := &GetFollowersResponse{} + if err := json.NewDecoder(res.Body).Decode(resData); err != nil { return nil, err } + + return resData, nil +} + +type ( + Subscriber struct { + BroadcasterID string `json:"broadcaster_id"` + BroadcasterLogin string `json:"broadcaster_login"` + BroadcasterName string `json:"broadcaster_name"` + + GifterID string `json:"gifter_id"` + GifterLogin string `json:"gifter_login"` + GifterName string `json:"gifter_name"` + + IsGift bool `json:"is_gift"` + Tier string `json:"tier"` + PlanName string `json:"plan_name"` + + UserID string `json:"user_id"` + UserLogin string `json:"user_login"` + UserName string `json:"user_name"` + } + + GetSubscribersResponse struct { + Data []Subscriber `json:"data"` + Pagination Pagination `json:"pagination"` + Total int `json:"total"` + Points int `json:"points"` + } +) +// Unfortunately, this function is very unreliable for pulling chronological +// subscription records. Twitch API does not currently provide a mechanism for +// fetching the latest subscriber; this will need to be tracked manually. +func (srv *Service) getSubscribers( + ctx context.Context, + broadcasterID string, + limit int, +) (*GetSubscribersResponse, error) { + client := srv.oauthConfig.Client(ctx, srv.oauthToken) + u, err := url.Parse(api.BASE_URL + "/subscriptions?" + url.Values{ + "broadcaster_id": []string{ broadcasterID }, + "first": []string { strconv.Itoa(limit) }, + }.Encode()) + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { return nil, 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 nil, err } + + if res.StatusCode != http.StatusOK { + body, _ := io.ReadAll(res.Body) + return nil, fmt.Errorf("%s: %s", res.Status, string(body)) + } + + resData := &GetSubscribersResponse{} + if err := json.NewDecoder(res.Body).Decode(resData); err != nil { return nil, err } + + return resData, nil +} diff --git a/twitch/public/labels.js b/twitch/public/labels.js index 7577aca..07bb8e1 100644 --- a/twitch/public/labels.js +++ b/twitch/public/labels.js @@ -40,8 +40,6 @@ async function sync() { console.error("Connection lost, or an error has occurred."); console.log("Attempting to reconnect..."); - // TODO: stop reconnecting after 10 failed attempts - setTimeout(() => { sync(); }, 1000); diff --git a/twitch/twitch.go b/twitch/twitch.go index 6d85221..844a8c7 100644 --- a/twitch/twitch.go +++ b/twitch/twitch.go @@ -1,7 +1,6 @@ package twitch import ( - "bytes" "context" "crypto/rand" "embed" @@ -52,15 +51,18 @@ type ( 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 + labels *twitchLabels } ) @@ -122,17 +124,14 @@ func New(ctx context.Context, opts ServiceOptions) (*Service, error) { 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, }, @@ -174,19 +173,6 @@ func New(ctx context.Context, opts ServiceOptions) (*Service, error) { } } - 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 } @@ -280,28 +266,47 @@ 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) + if err := srv.start(ctx); err != nil { + log.Printf("Failed to start Twitch service: %v", err) + } } } -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] - +func (srv *Service) start(ctx context.Context) error { + if res, err := srv.getUsers(nil, []string{ srv.channelName }); err == nil { + if len(res.Data) == 0 { + return fmt.Errorf("resolve username \"%s\", it may not exist?", srv.channelName) + } + srv.channelID = res.Data[0].ID + } else { return fmt.Errorf("fetch Twitch users: %v", err) } + if res, err := srv.getFollowers(ctx, srv.channelID, 1); err == nil { + if len(res.Data) > 0 { + srv.labels.LatestFollower.Text = res.Data[0].UserName + log.Printf("Loaded latest follower: %s", res.Data[0].UserName) + } + } else { + log.Printf("Failed to fetch latest follower: %v", err) + } + // TODO: build satellite service for offline-fetching subscribers and cheers + + 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)) + } + + srv.startWebsocketListener(ctx) + + return nil +} + +func (srv *Service) startWebsocketListener(ctx context.Context) { interrupt := make(chan os.Signal, 1) signal.Notify(interrupt, os.Interrupt) @@ -337,7 +342,7 @@ func (srv *Service) start(ctx context.Context) { srv.subscribeToDefaultEvents(ctx) - log.Printf("Connected to Twitch as %s.", srv.channelName) + log.Printf("Connected to Twitch for channel %s.", srv.channelName) } if message.Metadata.MessageType == string(api.NOTIFICATION) { @@ -353,334 +358,7 @@ func (srv *Service) start(ctx context.Context) { select { case err := <-failed: - log.Fatalf("Twitch error: %v", err) + log.Printf("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 -}