package twitch import ( "bytes" "context" "encoding/json" "fmt" "io" "log" "net/http" "net/url" "os" "path" "strconv" "codeberg.org/arimelody/ari-stream-tools/colour" "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 api.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("%sFollowed: %s%s", colour.Green, event.UserName, colour.Reset) srv.labels.LatestFollower.Text = event.UserName srv.labels.LatestFollower.C <- event.UserName srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("New Follow: %s", event.UserName), Type: SYSTEM_CHAT_FOLLOW, } 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 api.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("%sSubscribed: %s%s", colour.Purple, event.UserName, colour.Reset) srv.labels.LatestSubscriber.Text = event.UserName srv.labels.LatestSubscriber.C <- event.UserName srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("New Sub: %s", event.UserName), Type: SYSTEM_CHAT_SUBSCRIBE, } 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 api.CHANNEL_SUBSCRIPTION_MESSAGE: var event api.SubscribeMessageEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.SubscribeMessageEvent: %v", err) } log.Printf("%sResub: %s x%d%s", colour.Purple, event.UserName, event.CumulativeMonths, colour.Reset) if event.StreakMonths > 1 { log.Printf("%s is on a %d-month streak!", event.UserName, event.StreakMonths) } if len(event.Message.Text) > 0 { log.Printf("%s: %s", event.UserLogin, event.Message.Text) } srv.labels.LatestSubscriber.Text = event.UserName srv.labels.LatestSubscriber.C <- event.UserName srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("Resub: %s x%d", event.UserName, event.CumulativeMonths), Type: SYSTEM_CHAT_SUBSCRIBE, } 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 api.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%s cheered x%d bits:%s %s", colour.Cyan, event.UserName, event.Bits, colour.Reset, event.Message) srv.labels.LatestCheer.Text = event.UserName srv.labels.LatestCheer.C <- event.UserName srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("Cheer x%d %s: %s", event.Bits, event.UserName, event.Message), Type: SYSTEM_CHAT_CHEER, } 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 api.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("%sRAID: %s with %d viewers!%s", colour.Red, event.FromUserName, event.Viewers, colour.Reset) srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("%s raiding with %d viewers!", event.FromUserName, event.Viewers), Type: SYSTEM_CHAT_RAID, } case api.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 redeemed %s for %d channel points.", event.UserName, event.Reward.Title, event.Reward.Cost, ) srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("%s redeemed %s", event.UserName, event.Reward.Title), Type: SYSTEM_CHAT_POINT_REDEEM, } case api.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%s shouted out %s!%s", colour.Yellow, event.FromUserName, event.ToUserName, colour.Reset) srv.SystemChatC <- &systemChatMessage{ Text: fmt.Sprintf("Shoutout %s!", event.ToUserName), Type: SYSTEM_CHAT_SHOUTOUT, } case api.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 } var modifier ChatMessageModifier = CHAT_MODIFIER_NONE if event.MessageType == api.MESSAGE_CHANNEL_POINTS_HIGHLIGHTED { modifier = CHAT_MODIFIER_HIGHLIGHT log.Printf("Highlighted from %s: %s", event.ChatterName, event.Message.Text) } else { log.Printf("%s: %s", event.ChatterName, event.Message.Text) } srv.ChatC <- &chatMessage{ ID: event.MessageID, Username: event.ChatterName, Colour: event.Color, Badges: event.Badges, Text: event.Message.Text, Fragments: event.Message.Fragments, Modifier: modifier, } case api.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 deleted.", event.TargetLogin) srv.DeleteChatC <- &deleteChatMessage{ Username: "", MessageID: event.MessageID, } case api.CHANNEL_CHAT_NOTIFICATION: var event api.ChatNotificationEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.ChatNotificationEvent: %v", err) } switch event.NoticeType { case api.NOTICE_ANNOUNCEMENT: log.Printf("%sANNOUNCEMENT from %s:%s %s", colour.Yellow, event.ChatterName, colour.Reset, event.Message.Text) srv.ChatC <- &chatMessage{ ID: event.MessageID, Username: event.ChatterName, Colour: event.Color, Text: event.Message.Text, Fragments: event.Message.Fragments, Modifier: CHAT_MODIFIER_ANNOUNCEMENT, AnnouncementColour: event.Announcement.Color, } } case api.CHANNEL_CHAT_CLEAR: log.Printf("Chat cleared.") srv.DeleteChatC <- &deleteChatMessage{ Username: "*", MessageID: "*", } case api.CHANNEL_CHAT_CLEAR_USER_MESSAGES: var event api.ChatClearUserMessagesEvent err := json.Unmarshal(jsonData, &event) if err != nil { return fmt.Errorf("Failed to cast to api.ChatClearUserMessagesEvent: %v", err) } log.Printf("Messages from %s cleared.", event.TargetName) srv.DeleteChatC <- &deleteChatMessage{ Username: event.TargetName, MessageID: "*", } // TODO: polls (creation and results) default: log.Printf("Unhandled message type %s.", payload.Subscription.Type) } return nil } func (srv *Service) subscribeToEvent( ctx context.Context, subscriptionType api.SubscriptionType, version string, condition map[string]string, sessionID string, ) error { type ( Transport struct { Method string `json:"method"` SessionID string `json:"session_id"` } Request struct { Type api.SubscriptionType `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 } // Calls subscribeToEvent with helpful logging func (srv *Service) subscribeToEventHelper( ctx context.Context, subscriptionType api.SubscriptionType, version string, condition map[string]string, sessionID string, ) { if err := srv.subscribeToEvent(ctx, subscriptionType, version, condition, sessionID); err != nil { log.Printf("Failed to subscribe to %s: %v", subscriptionType, err) } } func (srv *Service) subscribeToDefaultEvents(ctx context.Context) { // Channel Updates srv.subscribeToEventHelper( ctx, api.CHANNEL_UPDATE, "2", map[string]string{ "broadcaster_user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Follows srv.subscribeToEventHelper( ctx, api.CHANNEL_FOLLOW, "2", map[string]string{ "broadcaster_user_id": srv.channelID, "moderator_user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Subscriptions srv.subscribeToEventHelper( ctx, api.CHANNEL_SUBSCRIBE, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Subscription Message srv.subscribeToEventHelper( ctx, api.CHANNEL_SUBSCRIPTION_MESSAGE, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Gift Subscriptions srv.subscribeToEventHelper( ctx, api.CHANNEL_SUBSCRIPTION_GIFT, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Cheer srv.subscribeToEventHelper( ctx, api.CHANNEL_CHEER, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Raid srv.subscribeToEventHelper( ctx, api.CHANNEL_RAID, "1", map[string]string{ "to_broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Channel Point Redeems srv.subscribeToEventHelper( ctx, api.CHANNEL_POINTS_CUSTOM_REWARD_REDEMPTION_ADD, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Shoutouts srv.subscribeToEventHelper( ctx, api.SHOUTOUT_CREATE, "1", map[string]string{ "broadcaster_user_id": srv.channelID, "moderator_user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Chat Messages srv.subscribeToEventHelper( ctx, api.CHANNEL_CHAT_MESSAGE, "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Chat Message Deletions srv.subscribeToEventHelper( ctx, api.CHANNEL_CHAT_MESSAGE_DELETE, "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Chat Notifications srv.subscribeToEventHelper( ctx, api.CHANNEL_CHAT_NOTIFICATION, "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Chat Clear srv.subscribeToEventHelper( ctx, api.CHANNEL_CHAT_CLEAR, "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Clear User Messages srv.subscribeToEventHelper( ctx, api.CHANNEL_CHAT_CLEAR_USER_MESSAGES, "1", map[string]string{ "broadcaster_user_id": srv.channelID, "user_id": srv.channelID, }, srv.eventSubSession.ID, ) // Polls (Starting) srv.subscribeToEventHelper( ctx, api.CHANNEL_POLL_BEGIN, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) // Polls (Ending) srv.subscribeToEventHelper( ctx, api.CHANNEL_POLL_END, "1", map[string]string{ "broadcaster_user_id": srv.channelID }, srv.eventSubSession.ID, ) } 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 } // 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. /* 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"` } ) 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 } */