Add EventSub client

config.yaml

@@ -1,4 +1,6 @@
 medgelabs:
+  server:
+    baseURL: "" # Note: can be overridden by the SERVER_BASE_URL env variable
   channelId: 62232210
   nick: medgelabs
   secretStore: env

config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	"medgebot/logger"
 	"os"
 
 	"github.com/pkg/errors"
@@ -15,7 +16,7 @@ type Config struct {
 	config  *viper.Viper
 }
 
-// Initialize configuration and read from config.yaml
+// New initializes configuration and read from config.yaml
 func New(channel string, configPath string) (Config, error) {
 	conf := viper.New()
 	conf.SetConfigName("config")
@@ -35,6 +36,18 @@ func New(channel string, configPath string) (Config, error) {
 	}, nil
 }
 
+// LocalServerBaseURL is the base URL that points to the embedded HTTP server
+func (c *Config) LocalServerBaseURL() string {
+	envOverride, present := os.LookupEnv("SERVER_BASE_URL")
+	if present {
+		logger.Info("Env override for SERVER_BASE_URL present")
+		return envOverride
+	}
+
+	value := c.config.GetString(c.key("server.baseURL"))
+	return value
+}
+
 // ChannelID returns the numeric ChannelID for the current broadcaster
 func (c *Config) ChannelID() string {
 	channelID := c.config.GetString(c.key("channelId"))
@@ -85,6 +98,11 @@ func (c *Config) TwitchToken() string {
 	return os.Getenv("TWITCH_TOKEN")
 }
 
+// ClientID if Store type is ENV
+func (c *Config) ClientID() string {
+	return os.Getenv("TWITCH_CLIENT_ID")
+}
+
 // Feature Flags - built as opt-in
 
 // GreeterEnabled checks the Greeter feature flag

eventsub/channelPoint.go

@@ -0,0 +1 @@
+package eventsub

eventsub/client.go

@@ -0,0 +1,133 @@
+package eventsub
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"medgebot/logger"
+	"net/http"
+
+	"github.com/buger/jsonparser"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+)
+
+const (
+	// JSON is a constant for the JSON ContentType header value
+	JSON = "application/json"
+
+	// ChallengeKey is the cache key for the challenge string received on Start
+	ChallengeKey = "challengeKey"
+)
+
+// Client is a client to the Twitch Client API that also handles message parsing for events
+type Client struct {
+	client *http.Client
+	secret string
+	Config
+}
+
+// Config is a input struct for the New() constructor
+// serverURL is the URL to the Twitch EventSub API
+// callbackURL is the local URL for the Bot which Twitch will send events to
+type Config struct {
+	ServerURL     string
+	CallbackURL   string
+	ClientID      string
+	AccessToken   string
+	BroadcasterID string
+}
+
+// New constructs a new Client to EventSub
+func New(config Config) Client {
+	secret := uuid.NewString()
+
+	return Client{
+		client: http.DefaultClient,
+		secret: secret,
+		Config: config,
+	}
+}
+
+// SubscriptionRequest represents a request to Twitch to create an EventSub subscription
+type SubscriptionRequest struct {
+	Type      string    `json:"type"`
+	Version   string    `json:"version"`
+	Condition Condition `json:"condition"`
+	Transport Transport `json:"transport"`
+}
+
+// Condition is the condition block of a SubscriptionRequest
+type Condition struct {
+	BroadcasterUserID string `json:"broadcaster_user_id"`
+}
+
+// Transport is the transport block of a SubscriptionRequest
+type Transport struct {
+	Method   string `json:"method"`
+	Callback string `json:"callback"`
+	Secret   string `json:"secret"`
+}
+
+// Start creates subscriptions to the desired events
+func (c *Client) Start() error {
+	if c.ServerURL == "" {
+		return errors.Errorf("serverURL must not be empty")
+	}
+	if c.CallbackURL == "" {
+		return errors.Errorf("callbackURL must not be empty")
+	}
+
+	jsonBody, err := json.Marshal(SubscriptionRequest{
+		Type:    "channel.channel_points_custom_reward_redemption.add",
+		Version: "1",
+		Condition: Condition{
+			BroadcasterUserID: c.BroadcasterID,
+		},
+		Transport: Transport{
+			Method:   "webhook",
+			Callback: c.CallbackURL,
+			Secret:   c.secret,
+		},
+	})
+	if err != nil {
+		return errors.Wrap(err, "Marshal subscription request failed")
+	}
+
+	url := fmt.Sprintf("%s/subscriptions", c.ServerURL)
+	req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(jsonBody))
+	if err != nil {
+		return errors.Wrap(err, "POST to create subscriptions failed")
+	}
+	req.Header.Add("Client-ID", c.ClientID)
+	req.Header.Add("Authorization", "Bearer "+c.AccessToken)
+	req.Header.Add("Content-Type", JSON)
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return errors.Wrap(err, "POST to create subscriptions failed")
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return errors.Wrap(err, "Read subscriptions response body")
+	}
+	defer resp.Body.Close()
+
+	challenge, err := jsonparser.GetString(body, "challenge")
+	if err != nil {
+		return errors.Wrap(err, "Get challenge key from body")
+	}
+
+	logger.Info("EventSub challenge key received: %s", challenge)
+
+	return nil
+}
+
+// Secret returns the generated secret used for Subscription initialization
+func (c *Client) Secret() string {
+	return c.secret
+}
+
+// TODO message parsing

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/buger/jsonparser v1.1.1
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
 	github.com/go-chi/chi/v5 v5.0.3
+	github.com/google/uuid v1.2.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/kr/pretty v0.2.1 // indirect
 	github.com/magiconair/properties v1.8.4 // indirect

go.sum

@@ -67,6 +67,8 @@ github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXi
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
+github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=

main.go

@@ -6,6 +6,7 @@ import (
 	"medgebot/bot"
 	"medgebot/cache"
 	"medgebot/config"
+	"medgebot/eventsub"
 	"medgebot/irc"
 	log "medgebot/logger"
 	"medgebot/pubsub"
@@ -67,7 +68,12 @@ func main() {
 		log.Fatal(err, "Create secret store")
 	}
 
-	password, err := store.TwitchToken()
+	clientID, err := store.ClientID()
+	if err != nil {
+		log.Fatal(err, "Get Client ID from store")
+	}
+
+	accessToken, err := store.TwitchToken()
 	if err != nil {
 		log.Fatal(err, "Get Twitch Token from store")
 	}
@@ -80,7 +86,7 @@ func main() {
 
 	ircConfig := irc.Config{
 		Nick:     nick,
-		Password: fmt.Sprintf("oauth:%s", password),
+		Password: fmt.Sprintf("oauth:%s", accessToken),
 		Channel:  channel,
 	}
 
@@ -115,7 +121,7 @@ func main() {
 			log.Fatal(err, "pubsub ws connect")
 		}
 
-		pubsub := pubsub.NewClient(pubSubWs, conf.ChannelID(), password)
+		pubsub := pubsub.NewClient(pubSubWs, conf.ChannelID(), accessToken)
 		pubSubWs.SetPostReconnectFunc(pubsub.Start)
 		pubsub.Start()
 		chatBot.RegisterClient(pubsub)
@@ -221,12 +227,27 @@ func main() {
 		log.Fatal(err, "bot connect")
 	}
 
+	// EventSub Client
+
+	// callbackURL should match the route found in server.routes()
+	callbackURL := conf.LocalServerBaseURL() + "/eventsub/callback"
+	eventSubClient := eventsub.New(eventsub.Config{
+		ServerURL:     "https://api.twitch.tv/helix/eventsub",
+		CallbackURL:   callbackURL,
+		BroadcasterID: conf.ChannelID(),
+		ClientID:      clientID,
+		AccessToken:   accessToken,
+	})
+	err = eventSubClient.Start()
+	if err != nil {
+		log.Fatal(err, "Connect to EventSub")
+	}
+
 	// Start HTTP server
 	// NOTE: Make sure the cache is the same as the Bot
 	debugClient := server.DebugClient{}
 	chatBot.RegisterClient(&debugClient)
-
-	srv := server.New(&chatBot, dataStore, &debugClient, metricsHTML, pollHTML)
+	srv := server.New(conf.LocalServerBaseURL(), &chatBot, dataStore, &debugClient, metricsHTML, pollHTML)
 	if err := http.ListenAndServe(fmt.Sprintf("%s:%s", listenAddr, listenPort), srv); err != nil {
 		log.Fatal(err, "start HTTP server")
 	}

secret/envStore.go

@@ -8,18 +8,31 @@ import (
 // environment variables
 type EnvStore struct {
 	twitchToken string
+	clientID    string
 }
 
-func NewEnvStore(twitchToken string) EnvStore {
+// NewEnvStore is an in-memory Secret store driven by ENV variables
+func NewEnvStore(twitchToken, clientID string) EnvStore {
 	return EnvStore{
 		twitchToken: twitchToken,
+		clientID:    clientID,
 	}
 }
 
+// TwitchToken returns the Twitch Access Token for the Bot
 func (s EnvStore) TwitchToken() (string, error) {
 	if s.twitchToken == "" {
 		return "", fmt.Errorf("ERROR: twitch token not in env")
 	}
 
 	return s.twitchToken, nil
 }
+
+// ClientID returns the Twitch OAuth ClientID used in API calls
+func (s EnvStore) ClientID() (string, error) {
+	if s.clientID == "" {
+		return "", fmt.Errorf("ERROR: client ID not in env")
+	}
+
+	return s.clientID, nil
+}

secret/store.go

@@ -3,4 +3,5 @@ package secret
 // Store fetches secrets needed for Bot integrations
 type Store interface {
 	TwitchToken() (string, error)
+	ClientID() (string, error)
 }

secret/storeFactory.go

@@ -9,6 +9,7 @@ import (
 )
 
 const (
+	// ENV gets secrets from environment variables
 	ENV = "env"
 )
 
@@ -22,7 +23,8 @@ func NewSecretStore(config config.Config) (Store, error) {
 	switch strings.ToLower(storeType) {
 	case ENV:
 		twitchToken := config.TwitchToken()
-		store := NewEnvStore(twitchToken)
+		clientID := config.ClientID()
+		store := NewEnvStore(twitchToken, clientID)
 		return &store, nil
 	default:
 		return nil, fmt.Errorf("Invalid storeType - %s. Valid values are: %v", storeType, []string{ENV})

server/eventSubHandler.go

@@ -0,0 +1,20 @@
+package server
+
+import (
+	"medgebot/eventsub"
+	"medgebot/logger"
+	"net/http"
+)
+
+func (s *Server) eventSubHandler(client eventsub.Client) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		// TODO switch for challenge key
+		challengeKey := client.Secret()
+		logger.Info("Responding to Event Sub challenge with %s", challengeKey)
+
+		w.WriteHeader(200)
+		w.Write([]byte(challengeKey))
+
+		// TODO parse events
+	}
+}