strimertul/twitch/client/client.go

package client

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"git.sr.ht/~ashkeel/strimertul/log"

	"git.sr.ht/~ashkeel/containers/sync"
	"github.com/nicklaw5/helix/v2"

	"git.sr.ht/~ashkeel/strimertul/database"
	"git.sr.ht/~ashkeel/strimertul/twitch"
	"git.sr.ht/~ashkeel/strimertul/twitch/alerts"
	"git.sr.ht/~ashkeel/strimertul/twitch/chat"
	"git.sr.ht/~ashkeel/strimertul/twitch/eventsub"
	"git.sr.ht/~ashkeel/strimertul/twitch/timers"
	"git.sr.ht/~ashkeel/strimertul/webserver"
)

type Manager struct {
	client *Client
}

func NewManager(ctx context.Context, db database.Database, server *webserver.WebServer) (*Manager, error) {
	logger := log.GetLogger(ctx)

	// Get Twitch config
	var config twitch.Config
	if err := db.GetJSON(twitch.ConfigKey, &config); err != nil {
		if !errors.Is(err, database.ErrEmptyKey) {
			return nil, fmt.Errorf("failed to get twitch config: %w", err)
		}
		config.Enabled = false
	}

	// Create new client
	clientContext, cancel := context.WithCancel(ctx)
	client, err := newClient(clientContext, config, db, server)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("failed to create twitch client: %w", err)
	}

	manager := &Manager{
		client: client,
	}

	// Listen for client config changes
	if err = db.SubscribeKeyContext(ctx, twitch.ConfigKey, func(value string) {
		var newConfig twitch.Config
		if err := json.Unmarshal([]byte(value), &newConfig); err != nil {
			logger.Error("Failed to decode Twitch integration config", log.Error(err))
			return
		}

		cancel()

		var updatedClient *Client
		clientContext, cancel = context.WithCancel(ctx)
		updatedClient, err = newClient(clientContext, newConfig, db, server)
		if err != nil {
			logger.Error("Could not create twitch client with new config, keeping old", log.Error(err))
			return
		}

		// New client works, replace old
		manager.client = updatedClient

		logger.Info("Reloaded/updated Twitch integration")
	}); err != nil {
		logger.Error("Could not setup twitch config reload subscription", log.Error(err))
	}

	return manager, nil
}

func (m *Manager) Client() *Client {
	return m.client
}

type Client struct {
	Config *sync.RWSync[twitch.Config]
	DB     database.Database
	API    *helix.Client
	User   helix.User
	Logger *slog.Logger

	Chat   *chat.Module
	Alerts *alerts.Module
	Timers *timers.Module

	eventSub *eventsub.Client
	server   *webserver.WebServer
	ctx      context.Context
	cancel   context.CancelFunc

	restart      chan bool
	streamOnline *sync.RWSync[bool]
}

func newClient(ctx context.Context, config twitch.Config, db database.Database, server *webserver.WebServer) (*Client, error) {
	// Create Twitch client
	client := &Client{
		Config:       sync.NewRWSync(config),
		DB:           db,
		Logger:       slog.With(slog.String("service", "twitch")),
		restart:      make(chan bool, 128),
		streamOnline: sync.NewRWSync(false),
		ctx:          ctx,
		server:       server,
	}

	if !config.Enabled {
		return client, nil
	}

	var err error
	client.API, err = twitch.GetHelixAPI(db)
	if err != nil {
		return nil, fmt.Errorf("failed to create twitch client: %w", err)
	}

	server.RegisterRoute(twitch.CallbackRoute, client)

	if userClient, err := twitch.GetUserClient(db, true); err == nil {
		users, err := userClient.GetUsers(&helix.UsersParams{})
		if err != nil {
			client.Logger.Error("Failed looking up user", log.Error(err))
		} else if len(users.Data.Users) < 1 {
			client.Logger.Error("No users found, please authenticate in Twitch configuration -> Events")
		} else {
			client.Logger.Info("Twitch user identified", slog.String("user", users.Data.Users[0].ID))
			client.User = users.Data.Users[0]
			client.eventSub, err = eventsub.Setup(ctx, userClient, client.User, db, client.Logger)
			if err != nil {
				client.Logger.Error("Failed to setup EventSub", log.Error(err))
			}

			tpl := client.GetTemplateEngine()
			client.Chat = chat.Setup(ctx, db, userClient, client.User, client.Logger, tpl)
			client.Alerts = alerts.Setup(ctx, db, client.Logger, tpl)
			client.Timers = timers.Setup(ctx, db, client.Logger)
		}
	} else {
		client.Logger.Warn("Twitch user not identified, this will break most features")
	}

	go client.runStatusPoll()

	go func() {
		<-ctx.Done()
		server.UnregisterRoute(twitch.CallbackRoute)
	}()

	return client, nil
}

func (c *Client) runStatusPoll() {
	c.Logger.Info("Started polling for stream status")
	for {
		// Make sure we're configured and connected properly first
		if !c.Config.Get().Enabled {
			continue
		}

		// Check if streamer is online, if possible
		func() {
			status, err := c.API.GetStreams(&helix.StreamsParams{
				UserIDs: []string{c.User.ID},
			})
			if err != nil {
				c.Logger.Error("Error checking stream status", log.Error(err))
				return
			}
			c.streamOnline.Set(len(status.Data.Streams) > 0)

			err = c.DB.PutJSON(twitch.StreamInfoKey, status.Data.Streams)
			if err != nil {
				c.Logger.Warn("Error saving stream info", log.Error(err))
			}
		}()

		// Wait for next poll (or cancellation)
		select {
		case <-c.ctx.Done():
			return
		case <-time.After(60 * time.Second):
		}
	}
}