Compare commits

...

2 commits

Author SHA1 Message Date
04ca14163e
more code tidy-up 2026-07-24 11:59:39 +01:00
2190cac343
code tidy-up 2026-07-24 11:02:32 +01:00
10 changed files with 506 additions and 380 deletions

View file

@ -7,10 +7,10 @@ tmp_dir = "tmp"
[build] [build]
args_bin = [] args_bin = []
bin = "./tmp/main" bin = "./tmp/stream-tools"
cmd = "go build -o ./tmp/main ." cmd = "go build -o ./tmp/stream-tools ./cmd/stream-tools"
delay = 1000 delay = 1000
entrypoint = ["./tmp/main"] entrypoint = ["./tmp/stream-tools"]
exclude_dir = ["assets", "tmp", "vendor", "testdata"] exclude_dir = ["assets", "tmp", "vendor", "testdata"]
exclude_file = [] exclude_file = []
exclude_regex = ["_test.go"] exclude_regex = ["_test.go"]

5
cmd/satellite/main.go Normal file
View file

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

View file

@ -43,10 +43,14 @@ func main() {
srv := gin.New() srv := gin.New()
srv.SetTrustedProxies([]string{"127.0.0.1", "::1"}) 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", 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) musicService, err := music.New(ctx)
if err != nil { if err != nil {

View file

@ -3,6 +3,7 @@ package learning
import ( import (
"context" "context"
"embed" "embed"
"fmt"
"io" "io"
"log" "log"
"net/http" "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{} titleText := []byte{}
titleUpdated := make(chan string) titleUpdated := make(chan string)
@ -45,10 +46,10 @@ func New(ctx context.Context, cfg ServiceConfig) *Service {
os.WriteFile(cfg.TitleFilePath, []byte("Untitled"), 0644) os.WriteFile(cfg.TitleFilePath, []byte("Untitled"), 0644)
titleText = []byte("Untitled") titleText = []byte("Untitled")
} else if err != nil { } else if err != nil {
log.Fatalf("Failed to stat title file: %v", err) return nil, fmt.Errorf("stat %s: %v", cfg.TitleFilePath, err)
} else { } else {
titleText, err = os.ReadFile(cfg.TitleFilePath) 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{ srv := &Service{
@ -60,7 +61,7 @@ func New(ctx context.Context, cfg ServiceConfig) *Service {
titleUpdatedBroadcast: broadcast.NewBroadcastChannel(ctx, titleUpdated), titleUpdatedBroadcast: broadcast.NewBroadcastChannel(ctx, titleUpdated),
} }
return srv return srv, nil
} }
func (srv *Service) BindRoutes(group *gin.RouterGroup) { func (srv *Service) BindRoutes(group *gin.RouterGroup) {

View file

@ -43,8 +43,6 @@ async function updateTitle() {
console.error("Connection lost, or an error has occurred."); console.error("Connection lost, or an error has occurred.");
console.log("Attempting to reconnect..."); console.log("Attempting to reconnect...");
// TODO: stop reconnecting after 10 failed attempts
setTimeout(() => { setTimeout(() => {
updateTitle(); updateTitle();
}, 1000); }, 1000);

View file

@ -225,7 +225,7 @@ func (srv *Service) Run(ctx context.Context) {
select { select {
case err := <-failed: case err := <-failed:
log.Fatalf("MPRIS/D-Bus error: %v", err) log.Printf("MPRIS/D-Bus error: %v", err)
case <-ctx.Done(): case <-ctx.Done():
} }
} }

View file

@ -37,8 +37,6 @@ async function sync() {
console.error("Connection lost, or an error has occurred."); console.error("Connection lost, or an error has occurred.");
console.log("Attempting to reconnect..."); console.log("Attempting to reconnect...");
// TODO: stop reconnecting after 10 failed attempts
setTimeout(() => { setTimeout(() => {
sync(); sync();
}, 1000); }, 1000);

444
twitch/api.go Normal file
View file

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

View file

@ -40,8 +40,6 @@ async function sync() {
console.error("Connection lost, or an error has occurred."); console.error("Connection lost, or an error has occurred.");
console.log("Attempting to reconnect..."); console.log("Attempting to reconnect...");
// TODO: stop reconnecting after 10 failed attempts
setTimeout(() => { setTimeout(() => {
sync(); sync();
}, 1000); }, 1000);

View file

@ -1,7 +1,6 @@
package twitch package twitch
import ( import (
"bytes"
"context" "context"
"crypto/rand" "crypto/rand"
"embed" "embed"
@ -52,15 +51,18 @@ type (
Service struct { Service struct {
port int16 port int16
labels *twitchLabels
channelName string channelName string
channelID string channelID string
clientID string clientID string
clientSecret string clientSecret string
oauthConfig *oauth2.Config oauthConfig *oauth2.Config
oauthState string oauthState string
oauthToken *oauth2.Token oauthToken *oauth2.Token
eventSubSession *api.EventSubSession eventSubSession *api.EventSubSession
labels *twitchLabels
} }
) )
@ -122,17 +124,14 @@ func New(ctx context.Context, opts ServiceOptions) (*Service, error) {
port: opts.Port, port: opts.Port,
labels: &twitchLabels{ labels: &twitchLabels{
LatestFollower: &twitchLabel{ LatestFollower: &twitchLabel{
Text: "some_follower",
C: latestFollowerC, C: latestFollowerC,
Broadcast: latestFollowerBroadcast, Broadcast: latestFollowerBroadcast,
}, },
LatestSubscriber: &twitchLabel{ LatestSubscriber: &twitchLabel{
Text: "some_subscriber",
C: latestSubscriberC, C: latestSubscriberC,
Broadcast: latestSubscriberBroadcast, Broadcast: latestSubscriberBroadcast,
}, },
LatestCheer: &twitchLabel{ LatestCheer: &twitchLabel{
Text: "some_cheer",
C: latestCheerC, C: latestCheerC,
Broadcast: latestCheerBroadcast, 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 return srv, nil
} }
@ -280,28 +266,47 @@ func (srv *Service) Run(ctx context.Context) {
if srv.oauthToken == nil || !srv.oauthToken.Valid() { if srv.oauthToken == nil || !srv.oauthToken.Valid() {
log.Printf("Log in with Twitch: http://localhost:%d/twitch/login", srv.port) log.Printf("Log in with Twitch: http://localhost:%d/twitch/login", srv.port)
} else { } 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) { func (srv *Service) start(ctx context.Context) error {
userIDs, err := srv.userIDsFromNames([]string{ srv.channelName }) if res, err := srv.getUsers(nil, []string{ srv.channelName }); err == nil {
if err != nil { if len(res.Data) == 0 {
log.Printf( return fmt.Errorf("resolve username \"%s\", it may not exist?", srv.channelName)
"Failed to resolve Twitch username \"%s\": %v", }
srv.channelName, srv.channelID = res.Data[0].ID
err, } else { return fmt.Errorf("fetch Twitch users: %v", err) }
)
return
}
if len(userIDs) == 0 {
fmt.Printf("Failed to resolve username \"%s\", it may not exist?", srv.channelName)
return
}
srv.channelID = userIDs[0]
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)
}
// 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) interrupt := make(chan os.Signal, 1)
signal.Notify(interrupt, os.Interrupt) signal.Notify(interrupt, os.Interrupt)
@ -337,7 +342,7 @@ func (srv *Service) start(ctx context.Context) {
srv.subscribeToDefaultEvents(ctx) 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) { if message.Metadata.MessageType == string(api.NOTIFICATION) {
@ -353,334 +358,7 @@ func (srv *Service) start(ctx context.Context) {
select { select {
case err := <-failed: case err := <-failed:
log.Fatalf("Twitch error: %v", err) log.Printf("Twitch error: %v", err)
case <-ctx.Done(): 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
}