stream-tools/twitch/twitch.go

package twitch

import (
	"bytes"
	"context"
	"crypto/rand"
	"embed"
	"encoding/json"
	"errors"
	"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
	}

	serviceConfig struct {
		ChannelName string `json:"channel_name"`
		ClientID string `json:"client_id"`
		ClientSecret string `json:"client_secret"`
	}

	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")
var CONFIG_FILENAME string = "twitch-config.json"
var CONFIG_PATH string = path.Join(DATA_PATH, CONFIG_FILENAME)

//go:embed public
var publicFS embed.FS
//go:embed pages
var pagesFS embed.FS

func New(ctx context.Context, opts ServiceOptions) (*Service, error) {
	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) }

	config := serviceConfig{}
	if configFile, err := os.OpenFile(CONFIG_PATH, os.O_CREATE | os.O_RDWR, 0600); err != nil {
		return nil, fmt.Errorf("open %s: %v", CONFIG_FILENAME, err)
	} else {
		defer configFile.Close()

		stat, err := configFile.Stat()
		if err != nil { return nil, fmt.Errorf("stat %s: %v", CONFIG_FILENAME, err) }

		if stat.Size() == 0 {
			enc := json.NewEncoder(configFile)
			enc.SetIndent("", "\t")
			if err := enc.Encode(&config); err != nil {
				return nil, fmt.Errorf("write %s: %v", CONFIG_FILENAME, err)
			}
			return nil, fmt.Errorf("Config file is empty: %s", CONFIG_PATH)
		} else if err := json.NewDecoder(configFile).Decode(&config); err != nil {
			return nil, fmt.Errorf("decode %s: %v", CONFIG_FILENAME, err)
		}
	}

	if len(config.ChannelName) == 0 { return nil, errors.New("config: channel_name cannot be empty") }
	if len(config.ClientID) == 0 { return nil, errors.New("config: client_id cannot be empty") }
	if len(config.ClientSecret) == 0 { return nil, errors.New("config: client_secret cannot be empty") }

	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: config.ChannelName,
		clientID: config.ClientID,
		clientSecret: config.ClientSecret,
		oauthConfig: &oauth2.Config{
			ClientID: config.ClientID,
			ClientSecret: config.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 authFile, err := os.OpenFile(path.Join(DATA_PATH, "twitch-auth.json"), os.O_RDONLY, 0600); err != nil {
		if !os.IsNotExist(err) {
			log.Fatalf("Failed to open twitch-auth.json: %v", err)
		}
	} else {
		defer authFile.Close()
		srv.oauthToken = &oauth2.Token{}
		err = json.NewDecoder(authFile).Decode(srv.oauthToken)
		if err != nil {
			log.Printf("Failed to read twitch-auth.json: %v", err)
		}
	}

	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("/label", 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
}