overhaul config, refactor twitch labels, init satellite

cmd/satellite/main.go

@@ -1,19 +1,74 @@
 package main
 
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"os/signal"
+	"syscall"
+
+	"codeberg.org/arimelody/ari-stream-tools/config"
+	"codeberg.org/arimelody/ari-stream-tools/twitch"
+	"github.com/gin-gonic/gin"
+)
+
 /*
  * -=[ SATELLITE ]=-
  *
  * a high-uptime companion service for stream-tools, tracking information that
  * would be unwieldy to track all the time on an end device.
- *                                                                                  
+ *
  * current use-case: fetching and storing the latest twitch subscribers and
  * cheers.
- * 
+ *
  * should be able to utilise a bunch of existing code under /twitch, then just
  * poke out an API for querying the information. SSE and websockets are
  * unnecessary; stream-tools already has this. we just need a "kick-start" payload.
  */
 
 func main() {
-	// TODO: build satellite service
+	failed := make(chan error)
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+
+	gin.SetMode(gin.ReleaseMode)
+
+	cfg, err := config.LoadConfig()
+	if err != nil { log.Fatalf("Failed to load config: %v", err) }
+
+	srv := gin.New()
+	// srv.Use(gin.Logger())
+	srv.SetTrustedProxies([]string{"127.0.0.1", "::1"})
+
+	twitchService, err := twitch.New(ctx, twitch.ServiceOptions{
+		Host: cfg.Host,
+		Port: cfg.Port,
+		ChannelName: cfg.Twitch.ChannelName,
+		ClientID: cfg.Twitch.ClientID,
+		ClientSecret: cfg.Twitch.ClientSecret,
+	})
+	if err != nil {
+		log.Printf("Failed to create Twitch service: %v", err)
+		srv.Group("twitch").Any("/*any", func(ctx *gin.Context) {
+			ctx.String(http.StatusServiceUnavailable, http.StatusText(http.StatusServiceUnavailable))
+		})
+	} else {
+		// satellite should work fine with the existing twitch service, though
+		// local stream-tools will need some kind of API override.
+		twitchService.BindRoutes(srv.Group("twitch"))
+		go twitchService.Run(ctx)
+	}
+
+	go func() {
+		log.Printf("Now serving at http://%s:%d\n", cfg.Host, cfg.Port)
+		failed <- srv.Run(fmt.Sprintf("%s:%d", cfg.Host, cfg.Port))
+	}()
+
+	select {
+	case err := <-failed:
+		log.Fatal(err)
+	case <-ctx.Done():
+		log.Print("Shutting down...")
+	}
 }

cmd/stream-tools/main.go

@@ -4,14 +4,12 @@ import (
 	"context"
 	"fmt"
 	"log"
-	"math"
 	"net/http"
-	"os"
 	"os/signal"
-	"strconv"
 	"strings"
 	"syscall"
 
+	"codeberg.org/arimelody/ari-stream-tools/config"
 	"codeberg.org/arimelody/ari-stream-tools/learning"
 	"codeberg.org/arimelody/ari-stream-tools/music"
 	"codeberg.org/arimelody/ari-stream-tools/static"
@@ -19,9 +17,6 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
-var DEFAULT_HOST string = "0.0.0.0"
-var DEFAULT_PORT int16 = 8080
-
 func main() {
 	failed := make(chan error)
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -29,19 +24,8 @@ func main() {
 
 	gin.SetMode(gin.ReleaseMode)
 
-	host := DEFAULT_HOST
+	cfg, err := config.LoadConfig()
-	port := DEFAULT_PORT
+	if err != nil { log.Fatalf("Failed to load config: %v", err) }
-
-	if len(os.Args) > 1 {
-		_port, err := strconv.Atoi(os.Args[1])
-		if _port > math.MaxUint16 {
-			log.Fatalf("Port must be between 1 and 65535: %d", _port)
-		}
-		port = int16(_port)
-		if err != nil {
-			log.Fatalf("Failed to parse port: %v", err)
-		}
-	}
 
 	srv := gin.New()
 	// srv.Use(gin.Logger())
@@ -70,7 +54,13 @@ func main() {
 		go musicService.Run(ctx)
 	}
 
-	twitchService, err := twitch.New(ctx, twitch.ServiceOptions{ Port: port })
+	twitchService, err := twitch.New(ctx, twitch.ServiceOptions{
+		Host: cfg.Host,
+		Port: cfg.Port,
+		ChannelName: cfg.Twitch.ChannelName,
+		ClientID: cfg.Twitch.ClientID,
+		ClientSecret: cfg.Twitch.ClientSecret,
+	})
 	if err != nil {
 		log.Printf("Failed to create Twitch service: %v", err)
 		srv.Group("twitch").Any("/*any", func(ctx *gin.Context) {
@@ -88,8 +78,8 @@ func main() {
 	})
 
 	go func() {
-		log.Printf("Now serving at http://%s:%d\n", host, port)
+		log.Printf("Now serving at http://%s:%d\n", cfg.Host, cfg.Port)
-		failed <- srv.Run(fmt.Sprintf("%s:%d", host, port))
+		failed <- srv.Run(fmt.Sprintf("%s:%d", cfg.Host, cfg.Port))
 	}()
 
 	select {

config/config.go

@@ -1,8 +1,25 @@
 package config
 
 import (
+	"fmt"
 	"os"
 	"path"
+
+	"github.com/pelletier/go-toml/v2"
+)
+
+type (
+	TwitchConfig struct {
+		ChannelName string `toml:"channel_name"`
+		ClientID string `toml:"client_id"`
+		ClientSecret string `toml:"client_secret"`
+	}
+
+	Config struct {
+		Host string `toml:"host"`
+		Port int16 `toml:"port"`
+		Twitch TwitchConfig `toml:"twitch"`
+	}
 )
 
 var CONFIG_DIR = func() string {
@@ -11,3 +28,40 @@ var CONFIG_DIR = func() string {
 	if err := os.MkdirAll(dir, 0750); err != nil { panic(err) }
 	return dir
 }()
+var configFilePath = path.Join(CONFIG_DIR, "config.toml")
+
+var DEFAULT_HOST string = "127.0.0.1"
+var DEFAULT_PORT int16 = 8080
+
+func DefaultConfig() *Config {
+	return &Config{
+		Host: DEFAULT_HOST,
+		Port: DEFAULT_PORT,
+		Twitch: TwitchConfig{
+			ChannelName: "",
+			ClientID: "",
+			ClientSecret: "",
+		},
+	}
+}
+
+func LoadConfig() (*Config, error) {
+	configFile, err := os.OpenFile(configFilePath, os.O_CREATE | os.O_RDWR, 0600)
+	if err != nil { return nil, fmt.Errorf("open %s: %v", configFilePath, err) }
+	defer configFile.Close()
+
+	config := DefaultConfig()
+	if stat, err := configFile.Stat(); err != nil {
+		return nil, fmt.Errorf("stat %s: %v", configFilePath, err)
+	} else if stat.Size() == 0 {
+		if err := toml.NewEncoder(configFile).Encode(config); err != nil {
+			return nil, fmt.Errorf("write %s: %v", configFilePath, err)
+		}
+		return config, nil
+	}
+
+	if err := toml.NewDecoder(configFile).Decode(config); err != nil {
+		return nil, fmt.Errorf("read %s: %v", configFilePath, err)
+	}
+	return config, nil
+}

twitch/oauth/oauth.go

@@ -1,77 +0,0 @@
-package oauth
-
-import (
-	"context"
-	"crypto/rand"
-	"fmt"
-	"log"
-	"net/http"
-	"sync"
-
-	"golang.org/x/oauth2"
-)
-
-func GenerateToken(
-	ctx *context.Context,
-	oauth2Config *oauth2.Config,
-) (*oauth2.Token, error) {
-	verifier := oauth2.GenerateVerifier()
-
-	state := rand.Text()
-
-	var token *oauth2.Token
-	wg := sync.WaitGroup{}
-
-	var server http.Server
-	server.Addr = "127.0.0.1"
-	server.Handler = http.HandlerFunc(
-		func(w http.ResponseWriter, r *http.Request) {
-			code := r.URL.Query().Get("code")
-			scope := r.URL.Query().Get("scope")
-			resState := r.URL.Query().Get("state")
-
-			if len(code) == 0 || len(scope) == 0 || resState != state {
-				http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
-				return
-			}
-
-			t, err := oauth2Config.Exchange(*ctx, code, oauth2.VerifierOption(verifier))
-			if err != nil {
-				log.Fatalf("Could not exchange OAuth2 code: %v", err)
-				http.Error( w,
-					fmt.Sprintf("Could not exchange OAuth2 code: %v", err),
-					http.StatusBadRequest,
-				)
-				return
-			}
-
-			token = t
-			http.Error(
-				w,
-				"Authentication successful! You may now close this tab.",
-				http.StatusOK,
-			)
-			if f, ok := w.(http.Flusher); ok {
-				f.Flush()
-			}
-
-			wg.Done()
-			server.Close()
-		},
-	)
-
-	url := oauth2Config.AuthCodeURL(
-		state,
-		oauth2.AccessTypeOffline,
-		oauth2.S256ChallengeOption(verifier),
-	)
-	log.Printf("Log in with Twitch: %s\n\n", url)
-
-	wg.Add(1)
-	if err := server.ListenAndServe(); err != http.ErrServerClosed {
-		return nil, fmt.Errorf("http: %v", err)
-	}
-	wg.Wait()
-
-	return token, nil
-}

twitch/public/labels.js

@@ -1,6 +1,6 @@
 const labelEl = document.getElementById("label");
 
-const selectedLabel = new URLSearchParams(location.search).get("l");
+const label = location.pathname.split("/twitch/labels/")[1];
 const labelLatestFollower   = "latest-follower"
 const labelLatestSubscriber = "latest-subscriber"
 const labelLatestCheer      = "latest-cheer"
@@ -11,14 +11,11 @@ const allowedLabels = [
 ]
 
 /**
- * @param {string} selectedLabel
+ * @param {string} label
  */
 async function sync() {
-    if (selectedLabel == null) {
+    if (label == null) {
-        const exampleUrl = 
+        const exampleUrl = `${location.protocol}//${location.host}/twitch/labels/${labelLatestFollower}`;
-            location.pathname +
-            "?l=" +
-            labelLatestFollower;
 
         labelEl.innerHTML =
             "Needs label to listen to, " +
@@ -29,7 +26,7 @@ async function sync() {
 
     let eventSource;
     try {
-        eventSource = new EventSource("/twitch/sse?l=" + selectedLabel);
+        eventSource = new EventSource("/twitch/sse/" + label);
     } catch (err) {
         console.error("Failed to connect to event source", err);
         return;
@@ -46,7 +43,7 @@ async function sync() {
     });
 
     eventSource.addEventListener("update", event => {
-        console.log(event.data);
+        console.log(`[${label}] ${event.data}`);
         labelEl.innerText = event.data;
     });
 

twitch/twitch.go

@@ -40,16 +40,16 @@ type (
 	}
 
 	ServiceOptions struct {
+		Host string
 		Port int16
-	}
 
-	serviceConfig struct {
+		ChannelName string
-		ChannelName string `json:"channel_name"`
+		ClientID string
-		ClientID string `json:"client_id"`
+		ClientSecret string
-		ClientSecret string `json:"client_secret"`
 	}
 
 	Service struct {
+		host string
 		port int16
 
 		channelName string
@@ -85,30 +85,9 @@ 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 len(opts.ChannelName) == 0 { return nil, errors.New("config: channel_name cannot be empty") }
-	if configFile, err := os.OpenFile(CONFIG_FILEPATH, os.O_CREATE | os.O_RDWR, 0600); err != nil {
+	if len(opts.ClientID) == 0 { return nil, errors.New("config: client_id cannot be empty") }
-		return nil, fmt.Errorf("open %s: %v", CONFIG_FILEPATH, err)
+	if len(opts.ClientSecret) == 0 { return nil, errors.New("config: client_secret cannot be empty") }
-	} else {
-		defer configFile.Close()
-
-		stat, err := configFile.Stat()
-		if err != nil { return nil, fmt.Errorf("stat %s: %v", CONFIG_FILEPATH, 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_FILEPATH, err)
-			}
-			return nil, fmt.Errorf("Config file is empty: %s", CONFIG_FILEPATH)
-		} else if err := json.NewDecoder(configFile).Decode(&config); err != nil {
-			return nil, fmt.Errorf("read %s: %v", CONFIG_FILEPATH, 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(
@@ -121,6 +100,7 @@ func New(ctx context.Context, opts ServiceOptions) (*Service, error) {
 		ctx, latestCheerC)
 
 	srv := &Service{
+		host: opts.Host,
 		port: opts.Port,
 		labels: &twitchLabels{
 			LatestFollower: &twitchLabel{
@@ -136,12 +116,12 @@ func New(ctx context.Context, opts ServiceOptions) (*Service, error) {
 				Broadcast: latestCheerBroadcast,
 			},
 		},
-		channelName: config.ChannelName,
+		channelName: opts.ChannelName,
-		clientID: config.ClientID,
+		clientID: opts.ClientID,
-		clientSecret: config.ClientSecret,
+		clientSecret: opts.ClientSecret,
 		oauthConfig: &oauth2.Config{
-			ClientID: config.ClientID,
+			ClientID: opts.ClientID,
-			ClientSecret: config.ClientSecret,
+			ClientSecret: opts.ClientSecret,
 			Endpoint: twitch.Endpoint,
 			Scopes: []string{
 				"moderator:read:followers",
@@ -156,7 +136,7 @@ func New(ctx context.Context, opts ServiceOptions) (*Service, error) {
 				"channel:read:hype_train",
 				"moderator:read:shoutouts",
 			},
-			RedirectURL: fmt.Sprintf("http://localhost:%d/twitch/auth", opts.Port),
+			RedirectURL: fmt.Sprintf("http://%s:%d/twitch/auth", opts.Host, opts.Port),
 		},
 	}
 
@@ -223,13 +203,13 @@ func (srv *Service) BindRoutes(group *gin.RouterGroup) {
 		go srv.start(ctx)
 	})
 
-	group.GET("/sse", func(ctx *gin.Context) {
+	group.GET("/sse/:label", func(ctx *gin.Context) {
 		ctx.Header("connection", "keep-alive")
 
-		listeningTo := ctx.Query("l")
+		labelParam := ctx.Param("label")
 
 		var label *twitchLabel
-		switch listeningTo {
+		switch labelParam {
 		case LABEL_LATEST_FOLLOWER:
 			label = srv.labels.LatestFollower
 		case LABEL_LATEST_SUBSCRIBER:
@@ -237,7 +217,7 @@ func (srv *Service) BindRoutes(group *gin.RouterGroup) {
 		case LABEL_LATEST_CHEER:
 			label = srv.labels.LatestCheer
 		default:
-			ctx.String(http.StatusBadRequest, "Unknown label %s", listeningTo)
+			ctx.String(http.StatusBadRequest, "Unknown label %s", labelParam)
 			return
 		}
 
@@ -257,14 +237,24 @@ func (srv *Service) BindRoutes(group *gin.RouterGroup) {
 		})
 	})
 
-	group.GET("/label", func(ctx *gin.Context) {
+	group.GET("/labels/:label", func(ctx *gin.Context) {
+		labelParam := ctx.Param("label")
+		switch labelParam {
+		case LABEL_LATEST_FOLLOWER:
+		case LABEL_LATEST_SUBSCRIBER:
+		case LABEL_LATEST_CHEER:
+		default:
+			http.NotFound(ctx.Writer, ctx.Request)
+			return
+		}
+
 		http.ServeFileFS(ctx.Writer, ctx.Request, pagesFS, "pages/labels.html")
 	})
 }
 
 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)
+		log.Printf("Log in with Twitch: http://%s:%d/twitch/login", srv.host, srv.port)
 	} else {
 		if err := srv.start(ctx); err != nil {
 			log.Printf("Failed to start Twitch service: %v", err)