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/main.go b/main.go index 57332e9..64791cc 100644 --- a/main.go +++ b/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/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/twitch/api.go b/twitch/api.go new file mode 100644 index 0000000..8094509 --- /dev/null +++ b/twitch/api.go @@ -0,0 +1,343 @@ +package twitch + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "path" + "strings" + + "codeberg.org/arimelody/ari-stream-tools/twitch/api" +) + +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 +} + +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) + } +} diff --git a/twitch/twitch.go b/twitch/twitch.go index 6d85221..27d3cc5 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 } ) @@ -299,9 +301,11 @@ func (srv *Service) start(ctx context.Context) { 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) @@ -353,334 +357,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 -}