feat: added twitch subscription events in api

event/event.go

@@ -52,6 +52,6 @@ func AddListeners(dc *discordgo.Session, t *twitchgo.Session, webChan chan struc
 	t.OnChannelMessage(twitch.MessageHandler)
 
 	addYouTubeListeners(dc)
-	addTwitchListeners(dc, t)
+	addTwitchListeners(dc, t, webChan)
 	addScheduledTriggers(dc, t, webChan)
 }

event/scheduledTriggers.go

@@ -28,22 +28,23 @@ import (
 )
 
 func addScheduledTriggers(dc *discordgo.Session, t *twitchgo.Session, webChan chan struct{}) {
-	go scheduleFunction(dc, t, 0, 0,
+	go scheduleFunction(dc, t, 0, 0, webChan,
 		adventcalendar.Midnight,
 	)
 
-	go scheduleFunction(dc, t, viper.GetInt("event.morning_hour"), viper.GetInt("event.morning_minute"),
+	go scheduleFunction(dc, t, viper.GetInt("event.morning_hour"), viper.GetInt("event.morning_minute"), webChan,
 		birthday.Check,
 		adventcalendar.Post,
 	)
 
 	go refreshYoutube(webChan)
 }
 
-func scheduleFunction(dc *discordgo.Session, t *twitchgo.Session, hour, min int, callbacks ...interface{}) {
+func scheduleFunction(dc *discordgo.Session, t *twitchgo.Session, hour, min int, webChan chan struct{}, callbacks ...interface{}) {
 	if len(callbacks) == 0 {
 		return
 	}
+	<-webChan
 	log.Printf("scheduled %d function(s) for %2d:%02d!", len(callbacks), hour, min)
 	time.Sleep(time.Second * 5)
 	for {

event/twitchUpdates.go

@@ -24,7 +24,7 @@ import (
 	"github.com/kesuaheli/twitchgo"
 )
 
-func addTwitchListeners(s *discordgo.Session, t *twitchgo.Session) {
+func addTwitchListeners(s *discordgo.Session, t *twitchgo.Session, webChan chan struct{}) {
 	webTwitch.SetDiscordSession(s)
 	webTwitch.SetTwitchSession(t)
 	webTwitch.SetDiscordChannelUpdateHandler(twitch.HandleChannelUpdate)
@@ -37,4 +37,9 @@ func addTwitchListeners(s *discordgo.Session, t *twitchgo.Session) {
 	if err != nil {
 		log.Printf("Error on subscribing to Twitch channels: %v", err)
 	}
+
+	go func() {
+		<-webChan
+		webTwitch.RefreshSubscriptions()
+	}()
 }

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/bwmarrin/discordgo v0.28.1
 	github.com/go-sql-driver/mysql v1.8.1
 	github.com/gorilla/mux v1.8.1
-	github.com/kesuaheli/twitchgo v1.0.0
+	github.com/kesuaheli/twitchgo v1.0.1-0.20241226023447-67ad1943851d
 	github.com/spf13/viper v1.19.0
 	golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f
 )

go.sum

@@ -20,8 +20,8 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/kesuaheli/twitchgo v1.0.0 h1:UhtK7Xj7nEJYX3XBRWsOL5x0W2P2dJf1GvaC29tjCJo=
-github.com/kesuaheli/twitchgo v1.0.0/go.mod h1:swIW1jJcFa4bWi/9JfUYH4Sf1kAvj+QUcaTPkce+6Ds=
+github.com/kesuaheli/twitchgo v1.0.1-0.20241226023447-67ad1943851d h1:CJvrH4jc//1f63SN8gTneONZA+n/xhkdmYAIX2/cT1k=
+github.com/kesuaheli/twitchgo v1.0.1-0.20241226023447-67ad1943851d/go.mod h1:swIW1jJcFa4bWi/9JfUYH4Sf1kAvj+QUcaTPkce+6Ds=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=

main.go

@@ -73,7 +73,8 @@ func main() {
 		log.Printf("Logged in to Discord as %s#%s\n", s.State.User.Username, s.State.User.Discriminator)
 	})
 
-	twitchBot := twitchgo.New(viper.GetString("twitch.clientID"), viper.GetString("twitch.clientSecret"), viper.GetString("twitch.token"))
+	twitchBot := twitchgo.New(viper.GetString("twitch.clientID"), viper.GetString("twitch.clientSecret"), viper.GetString("twitch.token")).
+		SetWebhookSecret(viper.GetString("twitch.webhookSecret"))
 
 	// adding listeners for events
 	event.AddListeners(discordBot, twitchBot, webChan)

webserver/main.go

@@ -34,9 +34,9 @@ func initHTTP() http.Handler {
 	r.NotFoundHandler = http.HandlerFunc(handle404)
 
 	r.HandleFunc("/favicon.ico", favicon)
-	r.HandleFunc("/api/twitch_pubsub", twitch.HandlePost).Methods(http.MethodPost)
-	r.HandleFunc("/api/yt_pubsubhubbub/", youtube.HandleGet).Methods("GET")
-	r.HandleFunc("/api/yt_pubsubhubbub/", youtube.HandlePost).Methods("POST")
+	r.HandleFunc(twitch.CALLBACKPATH, twitch.HandlePost).Methods(http.MethodPost)
+	r.HandleFunc(youtube.CALLBACKPATH, youtube.HandleGet).Methods(http.MethodGet)
+	r.HandleFunc(youtube.CALLBACKPATH, youtube.HandlePost).Methods(http.MethodPost)
 
 	return r
 }

webserver/twitch/api.go

@@ -128,7 +128,7 @@ func handleVerification(w http.ResponseWriter, _ *http.Request, rEvent rawEvent)
 		return
 	}
 	broadcaster := rEvent.Subscription.Condition["broadcaster_user_id"]
-	if subscribtions[broadcaster] {
+	if subscriptions[broadcaster] {
 		log.Printf("Declined verification for broadcaster '%s'!", broadcaster)
 		w.WriteHeader(http.StatusConflict)
 		w.Write([]byte("{\"conflict\":\"that broadcaster is not allowed\"}"))

webserver/twitch/discord.go

@@ -10,7 +10,7 @@ var tSession *twitchgo.Session
 var dcChannelUpdateHandler func(*discordgo.Session, *twitchgo.Session, *ChannelUpdateEvent)
 var dcStreamOnlineHandler func(*discordgo.Session, *twitchgo.Session, *StreamOnlineEvent)
 var dcStreamOfflineHandler func(*discordgo.Session, *twitchgo.Session, *StreamOfflineEvent)
-var subscribtions = make(map[string]bool)
+var subscriptions = make(map[string]bool)
 
 // SetDiscordSession sets the discordgo.Session to use for calling
 // event handlers.
@@ -45,17 +45,74 @@ func SetDiscordStreamOfflineHandler(f func(*discordgo.Session, *twitchgo.Session
 // SubscribeChannel subscribe to the event listener for new videos of
 // the given channel id.
 func SubscribeChannel(channelID string) {
-	if !subscribtions[channelID] {
-		subscribtions[channelID] = true
+	if !subscriptions[channelID] {
+		subscriptions[channelID] = true
 		log.Printf("subscribed '%s' for announcements", channelID)
 	}
 }
 
 // UnsubscribeChannel removes the given channel id from the
 // subscription list and no longer sends events.
 func UnsubscribeChannel(channelID string) {
-	if subscribtions[channelID] {
-		delete(subscribtions, channelID)
+	if subscriptions[channelID] {
+		delete(subscriptions, channelID)
 		log.Printf("unsubscribed '%s' from announcements", channelID)
 	}
 }
+
+// RefreshSubscriptions sends subscription requests for all registered channels
+// to subscribe stream events.
+func RefreshSubscriptions() {
+	var (
+		subscribedChannelUpdate = make(map[string]bool)
+		subscribedStreamOnline  = make(map[string]bool)
+		subscribedStreamOffline = make(map[string]bool)
+	)
+
+	getSubscriptions, err := tSession.GetSubscriptions(false)
+	if err != nil {
+		log.Printf("Error on getting subscribed channels for %s: %v", err, twitchgo.EventChannelUpdate)
+		return
+	}
+	for _, s := range getSubscriptions {
+		if s.Status == twitchgo.SubscriptionStatusWebhookCallbackVerificationFailed {
+			err = tSession.DeleteSubscription(s.ID)
+			if err != nil {
+				log.Printf("Error on deleting failed subscription '%s' for %s (%s): %v", s.Type, s.Condition["broadcaster_user_id"], s.ID, err)
+			} else {
+				log.Printf("Deleted failed subscription '%s' for %s", s.Type, s.Condition["broadcaster_user_id"])
+			}
+			continue
+		}
+
+		switch s.Type {
+		case twitchgo.EventChannelUpdate:
+			subscribedChannelUpdate[s.Condition["broadcaster_user_id"]] = true
+		case twitchgo.EventStreamOnline:
+			subscribedStreamOnline[s.Condition["broadcaster_user_id"]] = true
+		case twitchgo.EventStreamOffline:
+			subscribedStreamOffline[s.Condition["broadcaster_user_id"]] = true
+		}
+	}
+
+	for broadcasterID := range subscriptions {
+		if !subscribedChannelUpdate[broadcasterID] {
+			subscribe(broadcasterID, twitchgo.EventChannelUpdate)
+		}
+		if !subscribedStreamOnline[broadcasterID] {
+			subscribe(broadcasterID, twitchgo.EventStreamOnline)
+		}
+		if !subscribedStreamOffline[broadcasterID] {
+			subscribe(broadcasterID, twitchgo.EventStreamOffline)
+		}
+	}
+}
+
+func subscribe(broadcasterID string, event twitchgo.SubscriptionType) {
+	err := tSession.SubscribeToEvent(broadcasterID, CALLBACKURL, event)
+	if err != nil {
+		log.Printf("Error on subscribing '%s' for %s: %v", event, broadcasterID, err)
+	} else {
+		log.Printf("Requested subscription to '%s' for %s", event, broadcasterID)
+	}
+}

webserver/twitch/global.go → webserver/twitch/twitch.go

@@ -4,6 +4,13 @@ import (
 	"time"
 )
 
+const (
+	// CALLBACKPATH is the endpoint that twitch will call to send events.
+	CALLBACKPATH string = "/api/twitch_pubsub"
+	// CALLBACKURL is the full URL to the CALLBACKPATH.
+	CALLBACKURL string = "https://webhook.cake4everyone.de" + CALLBACKPATH
+)
+
 // Subscription represents a single subscription to an event.
 type Subscription struct {
 	// ID is the unique identifier for this subscription.

webserver/youtube/discord.go

@@ -70,7 +70,7 @@ func refreshSubscription(id string) {
 	reqURL := "https://pubsubhubbub.appspot.com/subscribe"
 
 	form := url.Values{}
-	form.Set("hub.callback", "https://webhook.cake4everyone.de/api/yt_pubsubhubbub/")
+	form.Set("hub.callback", CALLBACKURL)
 	form.Set("hub.topic", "https://www.youtube.com/xml/feeds/videos.xml?channel_id="+id)
 	form.Set("hub.verify", "sync")
 	form.Set("hub.mode", "subscribe")

webserver/youtube/youtube.go

@@ -179,6 +179,10 @@ func (c Category) String() string {
 
 const (
 	youtubeAPIBaseURL string = "https://youtube.googleapis.com/youtube/v3"
+	// CALLBACKPATH is the endpoint that youtube will call to send events.
+	CALLBACKPATH string = "/api/yt_pubsubhubbub/"
+	// CALLBACKURL is the full URL to the CALLBACKPATH.
+	CALLBACKURL string = "https://webhook.cake4everyone.de" + CALLBACKPATH
 )
 
 // checkVideo checks if a video really is from the provided channel