t5de-proxy/internal/hub/hub.go

// Package hub connects the proxy to T5DE's hub (git.t5de.dev/t5de/t5de-hub): signing in with a code the person
// enters on the hub's website, and a live connection reporting which IMVU account the client is using and the
// community products it wears.
package hub

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/coder/websocket"
)

// Item is a community product worn.
type Item struct {
	ID   string `json:"id"`
	Name string `json:"name"`
}

// Saved is the proxy's sign-in, in <data dir>/hub.json. The refresh token is encrypted for the Windows user.
type Saved struct {
	URL          string `json:"url"`
	RefreshToken string `json:"-"`
	Protected    []byte `json:"refresh_token"`
}

// ErrSignedOut means there's no sign-in, or the hub ended it.
var ErrSignedOut = errors.New("not signed in to the hub")

func Load(path string) (*Saved, error) {
	b, err := os.ReadFile(path)
	if errors.Is(err, os.ErrNotExist) {
		return nil, ErrSignedOut
	}
	if err != nil {
		return nil, err
	}
	var s Saved
	if err := json.Unmarshal(b, &s); err != nil {
		return nil, err
	}
	tok, err := unprotect(s.Protected)
	if err != nil {
		return nil, fmt.Errorf("reading the hub sign-in: %w", err)
	}
	s.RefreshToken = string(tok)
	return &s, nil
}

func (s *Saved) Save(path string) error {
	p, err := protect([]byte(s.RefreshToken))
	if err != nil {
		return err
	}
	s.Protected = p
	b, err := json.MarshalIndent(s, "", "  ")
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, b, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

type tokens struct {
	AccessToken  string `json:"access_token"`
	RefreshToken string `json:"refresh_token"`
	Error        string `json:"error"`
}

func post(ctx context.Context, endpoint string, form url.Values) (*tokens, int, error) {
	req, err := http.NewRequestWithContext(ctx, "POST", endpoint, strings.NewReader(form.Encode()))
	if err != nil {
		return nil, 0, err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	req.Header.Set("User-Agent", "t5de-proxy")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, 0, err
	}
	defer resp.Body.Close()
	var t tokens
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err := json.Unmarshal(body, &t); err != nil {
		return nil, resp.StatusCode, fmt.Errorf("the hub answered %d", resp.StatusCode)
	}
	return &t, resp.StatusCode, nil
}

// SignIn signs the proxy in: it gets a code from the hub, shows it (show gets the code and the page to enter it
// on), and waits until the person has entered it.
func SignIn(ctx context.Context, hubURL string, show func(code, page string)) (*Saved, error) {
	hubURL = strings.TrimSuffix(hubURL, "/")
	req, _ := http.NewRequestWithContext(ctx, "POST", hubURL+"/api/v1/device/code", nil)
	req.Header.Set("User-Agent", "t5de-proxy")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	var device struct {
		DeviceCode string `json:"device_code"`
		UserCode   string `json:"user_code"`
		Complete   string `json:"verification_uri_complete"`
		Interval   int    `json:"interval"`
		ExpiresIn  int    `json:"expires_in"`
	}
	err = json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&device)
	resp.Body.Close()
	if err != nil || device.DeviceCode == "" {
		return nil, fmt.Errorf("the hub didn't give a sign-in code (%d)", resp.StatusCode)
	}
	show(device.UserCode, device.Complete)
	interval := time.Duration(max(device.Interval, 2)) * time.Second
	deadline := time.Now().Add(time.Duration(device.ExpiresIn) * time.Second)
	for time.Now().Before(deadline) {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(interval):
		}
		t, _, err := post(ctx, hubURL+"/api/v1/device/token", url.Values{"device_code": {device.DeviceCode}})
		if err != nil {
			continue
		}
		switch t.Error {
		case "":
			return &Saved{URL: hubURL, RefreshToken: t.RefreshToken}, nil
		case "authorization_pending", "slow_down":
			continue
		default:
			return nil, fmt.Errorf("sign-in failed: %s", t.Error)
		}
	}
	return nil, errors.New("the code expired")
}

// Client is the live connection to the hub.
type Client struct {
	path string
	log  *log.Logger

	mu     sync.Mutex
	saved  *Saved
	imvuID int64
	outfit []Item
	wake   chan struct{}
}

// New makes a client for the sign-in saved at path.
func New(path string, saved *Saved, logger *log.Logger) *Client {
	return &Client{path: path, saved: saved, log: logger, wake: make(chan struct{}, 1)}
}

// Session reports the IMVU account the client signed in as.
func (c *Client) Session(imvuID int64) {
	c.mu.Lock()
	changed := c.imvuID != imvuID
	c.imvuID = imvuID
	if changed {
		c.outfit = nil
	}
	c.mu.Unlock()
	if changed {
		c.poke()
	}
}

// Outfit reports the community products the account wears.
func (c *Client) Outfit(imvuID int64, items []Item) {
	c.mu.Lock()
	if c.imvuID != imvuID {
		c.mu.Unlock()
		return
	}
	c.outfit = append([]Item(nil), items...)
	c.mu.Unlock()
	c.poke()
}

func (c *Client) poke() {
	select {
	case c.wake <- struct{}{}:
	default:
	}
}

// Run keeps the connection up until ctx ends or the hub signs the proxy out.
func (c *Client) Run(ctx context.Context) {
	backoff := time.Second
	for ctx.Err() == nil {
		start := time.Now()
		err := c.connect(ctx)
		if errors.Is(err, ErrSignedOut) {
			c.log.Printf("hub: signed out (%v); run t5de-proxy signin again", err)
			return
		}
		if ctx.Err() != nil {
			return
		}
		if time.Since(start) > time.Minute {
			backoff = time.Second
		}
		c.log.Printf("hub: disconnected (%v); retrying in %s", err, backoff)
		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff):
		}
		backoff = min(backoff*2, 2*time.Minute)
	}
}

// accessToken trades the refresh token for an access token (and a new refresh token, saved).
func (c *Client) accessToken(ctx context.Context) (string, error) {
	c.mu.Lock()
	saved := *c.saved
	c.mu.Unlock()
	t, status, err := post(ctx, saved.URL+"/api/v1/token/refresh", url.Values{"refresh_token": {saved.RefreshToken}})
	if err != nil {
		return "", err
	}
	if status == http.StatusUnauthorized || status == http.StatusForbidden {
		return "", fmt.Errorf("%w: %s", ErrSignedOut, t.Error)
	}
	if t.Error != "" || t.AccessToken == "" {
		return "", fmt.Errorf("the hub answered %d %s", status, t.Error)
	}
	saved.RefreshToken = t.RefreshToken
	if err := saved.Save(c.path); err != nil {
		return "", err
	}
	c.mu.Lock()
	c.saved = &saved
	c.mu.Unlock()
	return t.AccessToken, nil
}

func (c *Client) connect(ctx context.Context) error {
	access, err := c.accessToken(ctx)
	if err != nil {
		return err
	}
	wsURL := strings.Replace(strings.Replace(c.saved.URL, "https://", "wss://", 1), "http://", "ws://", 1) + "/api/v1/realtime"
	conn, _, err := websocket.Dial(ctx, wsURL, &websocket.DialOptions{HTTPHeader: http.Header{
		"Authorization": {"Bearer " + access}, "User-Agent": {"t5de-proxy"}}})
	if err != nil {
		return err
	}
	defer conn.CloseNow()
	c.log.Printf("hub: connected to %s", c.saved.URL)

	errs := make(chan error, 1)
	go func() {
		for {
			_, data, err := conn.Read(ctx)
			if err != nil {
				errs <- err
				return
			}
			var m struct{ Type, Message string }
			if json.Unmarshal(data, &m) == nil && m.Type == "error" {
				c.log.Printf("hub: %s", m.Message)
			}
		}
	}()
	send := func(v any) error {
		b, _ := json.Marshal(v)
		wctx, cancel := context.WithTimeout(ctx, 10*time.Second)
		defer cancel()
		return conn.Write(wctx, websocket.MessageText, b)
	}
	// Access tokens are short-lived; reconnecting with a fresh one before it expires keeps the hub's check current.
	renew := time.After(12 * time.Minute)
	sentID, sentOutfit := int64(-1), ""
	for {
		c.mu.Lock()
		id, outfit := c.imvuID, append([]Item(nil), c.outfit...)
		c.mu.Unlock()
		if id != sentID && id != 0 {
			if err := send(map[string]any{"type": "session", "imvu_id": id}); err != nil {
				return err
			}
			sentID, sentOutfit = id, ""
		}
		if b, _ := json.Marshal(outfit); id != 0 && string(b) != sentOutfit {
			if err := send(map[string]any{"type": "outfit", "items": outfit}); err != nil {
				return err
			}
			sentOutfit = string(b)
		}
		select {
		case <-ctx.Done():
			conn.Close(websocket.StatusNormalClosure, "")
			return ctx.Err()
		case err := <-errs:
			return err
		case <-renew:
			conn.Close(websocket.StatusNormalClosure, "renewing")
			return errors.New("renewing the access token")
		case <-c.wake:
		}
	}
}