more code tidy-up

.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"]

cmd/satellite/main.go

@@ -0,0 +1,5 @@
+package main
+
+func main() {
+	// TODO: build satellite service for offline-fetching subscribers and cheers
+}

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",
 	})
+	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 {

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) {

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);

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():
 	}
 }

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);

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
+}

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);

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
+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)
 		}
-	if len(userIDs) == 0 {
-		fmt.Printf("Failed to resolve username \"%s\", it may not exist?", srv.channelName)
-		return
+		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)
 	}
-	srv.channelID = userIDs[0]
 
+	// 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
-}