stream-tools/twitch/api.go

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.ChatterLogin, event.Message.Text)
		} else {
			log.Printf("%s: %s", event.ChatterLogin, event.Message.Text)
		}

		srv.ChatC <- &chatMessage{
			ID: event.MessageID,
			Username: event.ChatterLogin,
			Colour: event.Color,
			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
}
*/