backend code

server/twitch/auth.go

@@ -0,0 +1,75 @@
+package twitch
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+const authURL = "https://id.twitch.tv/oauth2/token"
+
+type AuthResponse struct {
+	AccessToken string `json:"access_token"`
+	ExpiresIn   int    `json:"expires_in"`
+	TokenType   string `json:"token_type"`
+}
+
+type AccessToken struct {
+	Token  string
+	Expiry time.Time
+}
+
+type AuthenticationManager struct {
+	clientId     string
+	clientSecret string
+	accesToken   *AccessToken
+}
+
+func NewAuthenticationManager(clientId, clientSecret string) *AuthenticationManager {
+	return &AuthenticationManager{
+		clientId:     clientId,
+		clientSecret: clientSecret,
+		accesToken:   &AccessToken{},
+	}
+}
+
+func (a *AuthenticationManager) GetAccessToken() (*AccessToken, error) {
+	if a.accesToken != nil && a.accesToken.Expiry.After(time.Now()) {
+		return a.accesToken, nil
+	}
+
+	data := url.Values{}
+	data.Set("client_id", a.clientId)
+	data.Set("client_secret", a.clientSecret)
+	data.Set("grant_type", "client_credentials")
+
+	resp, err := http.PostForm(authURL, data)
+	if err != nil {
+		return nil, fmt.Errorf("errore richiesta token: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("status non OK: %s", resp.Status)
+	}
+
+	var auth AuthResponse
+	if err := json.NewDecoder(resp.Body).Decode(&auth); err != nil {
+		return nil, fmt.Errorf("errore decoding JSON: %w", err)
+	}
+
+	token := &AccessToken{
+		Token:  auth.AccessToken,
+		Expiry: time.Now().Add(time.Duration(auth.ExpiresIn) * time.Second),
+	}
+
+	a.accesToken = token
+
+	return token, nil
+}
+
+func (a *AuthenticationManager) GetClientId() string {
+	return a.clientId
+}

server/twitch/client.go

@@ -0,0 +1,86 @@
+package twitch
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"time"
+)
+
+const twitchAPIURL = "https://api.twitch.tv/helix"
+
+type Client struct {
+	authenticationManager AuthenticationManager
+}
+
+func NewTwitchClient(am *AuthenticationManager) *Client {
+	return &Client{
+		authenticationManager: *am,
+	}
+}
+
+type streamResp struct {
+	Data []struct {
+		ID        string `json:"id"`
+		UserName  string `json:"user_name"`
+		Title     string `json:"title"`
+		GameName  string `json:"game_name"`
+		StartedAt string `json:"started_at"`
+	} `json:"data"`
+}
+
+func (c *Client) doRequest(endpoint string, params map[string]string) ([]byte, error) {
+	token, err := c.authenticationManager.GetAccessToken()
+	if err != nil {
+		return nil, err
+	}
+
+	reqURL := twitchAPIURL + endpoint
+	req, err := http.NewRequest("GET", reqURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	q := req.URL.Query()
+	for k, v := range params {
+		q.Set(k, v)
+	}
+	req.URL.RawQuery = q.Encode()
+
+	req.Header.Set("Client-Id", c.authenticationManager.GetClientId())
+	req.Header.Set("Authorization", "Bearer "+token.Token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	return io.ReadAll(resp.Body)
+}
+
+func (c *Client) PollStream(channel string, liveChannel chan<- *StreamInfo) error {
+	body, err := c.doRequest("/streams", map[string]string{"user_login": channel})
+	if err != nil {
+		return err
+	}
+
+	var sr streamResp
+	if err := json.Unmarshal(body, &sr); err != nil {
+		return err
+	}
+
+	s := sr.Data[0]
+	started, _ := time.Parse(time.RFC3339, s.StartedAt)
+
+	liveChannel <- &StreamInfo{
+		ID:        s.ID,
+		UserName:  s.UserName,
+		Title:     s.Title,
+		GameName:  s.GameName,
+		StartedAt: started,
+		IsLive:    true,
+	}
+
+	return nil
+}

server/twitch/monitor.go

@@ -0,0 +1,142 @@
+package twitch
+
+import (
+	"context"
+	"encoding/gob"
+	"fmt"
+	"iter"
+	"log/slog"
+	"maps"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/config"
+	"github.com/marcopiovanello/yt-dlp-web-ui/v3/server/internal"
+)
+
+type Monitor struct {
+	liveChannel           chan *StreamInfo
+	monitored             map[string]*Client
+	lastState             map[string]bool
+	mu                    sync.RWMutex
+	authenticationManager *AuthenticationManager
+}
+
+func NewMonitor(authenticationManager *AuthenticationManager) *Monitor {
+	return &Monitor{
+		liveChannel:           make(chan *StreamInfo, 16),
+		monitored:             make(map[string]*Client),
+		lastState:             make(map[string]bool),
+		authenticationManager: authenticationManager,
+	}
+}
+
+func (m *Monitor) Add(user string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.monitored[user] = NewTwitchClient(m.authenticationManager)
+	slog.Info("added user to twitch monitor", slog.String("user", user))
+}
+
+func (m *Monitor) Monitor(ctx context.Context, interval time.Duration, handler func(url string) error) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			m.mu.RLock()
+			for user, client := range m.monitored {
+				u := user
+				c := client
+
+				go func() {
+					if err := c.PollStream(u, m.liveChannel); err != nil {
+						slog.Error("polling failed", slog.String("user", u), slog.Any("err", err))
+					}
+				}()
+			}
+			m.mu.RUnlock()
+
+		case stream := <-m.liveChannel:
+			wasLive := m.lastState[stream.UserName]
+			if stream.IsLive && !wasLive {
+				slog.Info("stream went live", slog.String("user", stream.UserName))
+				if err := handler(fmt.Sprintf("https://www.twitch.tv/%s", stream.UserName)); err != nil {
+					slog.Error("handler failed", slog.String("user", stream.UserName), slog.Any("err", err))
+				}
+			}
+			m.lastState[stream.UserName] = stream.IsLive
+
+		case <-ctx.Done():
+			slog.Info("stopping twitch monitor")
+			return
+		}
+	}
+}
+
+func (m *Monitor) GetMonitoredUsers() iter.Seq[string] {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return maps.Keys(m.monitored)
+}
+
+func DEFAULT_DOWNLOAD_HANDLER(db *internal.MemoryDB, mq *internal.MessageQueue) func(url string) error {
+	return func(url string) error {
+		p := &internal.Process{
+			Url:        url,
+			Livestream: true,
+			Params:     []string{"--downloader", "ffmpeg", "--no-part"},
+		}
+		db.Set(p)
+		mq.Publish(p)
+		return nil
+	}
+}
+
+func (m *Monitor) Persist() error {
+	filename := filepath.Join(config.Instance().SessionFilePath, "twitch-monitor.dat")
+
+	f, err := os.Create(filename)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	enc := gob.NewEncoder(f)
+	users := make([]string, 0, len(m.monitored))
+
+	for user := range m.monitored {
+		users = append(users, user)
+	}
+
+	return enc.Encode(users)
+}
+
+func (m *Monitor) Restore() error {
+	filename := filepath.Join(config.Instance().SessionFilePath, "twitch-monitor.dat")
+
+	f, err := os.Open(filename)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+	defer f.Close()
+
+	dec := gob.NewDecoder(f)
+	var users []string
+	if err := dec.Decode(&users); err != nil {
+		return err
+	}
+
+	m.monitored = make(map[string]*Client)
+	for _, user := range users {
+		m.monitored[user] = NewTwitchClient(m.authenticationManager)
+	}
+
+	return nil
+}

server/twitch/rest.go

@@ -0,0 +1,28 @@
+package twitch
+
+import (
+	"encoding/json"
+	"net/http"
+)
+
+type addUserReq struct {
+	User string `json:"user"`
+}
+
+func MonitorUserHandler(m *Monitor) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var req addUserReq
+
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		m.Add(req.User)
+
+		if err := json.NewEncoder(w).Encode("ok"); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	}
+}

server/twitch/types.go

@@ -0,0 +1,20 @@
+package twitch
+
+import "time"
+
+type StreamInfo struct {
+	ID        string
+	UserName  string
+	Title     string
+	GameName  string
+	StartedAt time.Time
+	IsLive    bool
+}
+
+type VodInfo struct {
+	ID        string
+	Title     string
+	URL       string
+	Duration  string
+	CreatedAt time.Time
+}