t5de-hub/internal/realtime/realtime.go

// Package realtime is the hub's live connection with signed-in proxies: which IMVU account each client is using
// (presence), and the community products it's wearing. Later, room channels.
//
// Protocol (JSON text messages over a WebSocket at /api/v1/realtime, authorized with the proxy's access token):
//
//	proxy -> hub  {"type": "session", "imvu_id": 123}             the client signed in as this IMVU account
//	proxy -> hub  {"type": "outfit", "items": [{"id", "name"}]}   the community products it wears now
//	hub -> proxy  {"type": "welcome"} | {"type": "error", "message": "..."}
package realtime

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"sort"
	"sync"
	"time"

	"github.com/coder/websocket"
)

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

// Online is an IMVU account in use in a T5DE client now.
type Online struct {
	PersonID int64
	IMVUID   int64
	Since    time.Time
	Outfit   []Item
}

// Owners tells the hub which person an IMVU account is linked to.
type Owners interface {
	LinkOwner(ctx context.Context, imvuID int64) (personID int64, ok bool, err error)
}

// Hub holds the connected proxies.
type Hub struct {
	owners Owners
	log    *log.Logger

	mu       sync.Mutex
	sessions map[*session]bool
	changed  chan struct{} // closed and replaced on every change, for live views
}

type session struct {
	personID int64
	conn     *websocket.Conn
	imvuID   int64
	since    time.Time
	outfit   []Item
}

func New(owners Owners, logger *log.Logger) *Hub {
	return &Hub{owners: owners, log: logger, sessions: map[*session]bool{}, changed: make(chan struct{})}
}

// Changed is closed the next time anyone's presence or outfit changes.
func (h *Hub) Changed() <-chan struct{} {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.changed
}

func (h *Hub) notifyLocked() {
	close(h.changed)
	h.changed = make(chan struct{})
}

// Online lists the IMVU accounts in use now, one entry each (the most recent session), by IMVU ID.
func (h *Hub) Online() []Online {
	h.mu.Lock()
	defer h.mu.Unlock()
	byID := map[int64]Online{}
	for s := range h.sessions {
		if s.imvuID == 0 {
			continue
		}
		if prev, ok := byID[s.imvuID]; ok && prev.Since.After(s.since) {
			continue
		}
		byID[s.imvuID] = Online{PersonID: s.personID, IMVUID: s.imvuID, Since: s.since, Outfit: append([]Item(nil), s.outfit...)}
	}
	out := make([]Online, 0, len(byID))
	for _, o := range byID {
		out = append(out, o)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Since.Before(out[j].Since) })
	return out
}

// Connections counts connected proxies (signed in or not to an IMVU account yet).
func (h *Hub) Connections() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.sessions)
}

// Disconnect closes a person's connections (when they're banned or sign their proxies out).
func (h *Hub) Disconnect(personID int64) {
	h.mu.Lock()
	var conns []*websocket.Conn
	for s := range h.sessions {
		if s.personID == personID {
			conns = append(conns, s.conn)
		}
	}
	h.mu.Unlock()
	for _, c := range conns {
		c.Close(websocket.StatusPolicyViolation, "signed out")
	}
}

type message struct {
	Type    string `json:"type"`
	IMVUID  int64  `json:"imvu_id,omitempty"`
	Items   []Item `json:"items,omitempty"`
	Message string `json:"message,omitempty"`
}

// Serve runs one proxy's connection, already authorized as personID, until it closes.
func (h *Hub) Serve(w http.ResponseWriter, r *http.Request, personID int64) {
	conn, err := websocket.Accept(w, r, nil)
	if err != nil {
		return
	}
	conn.SetReadLimit(64 << 10)
	s := &session{personID: personID, conn: conn, since: time.Now()}
	h.mu.Lock()
	h.sessions[s] = true
	h.mu.Unlock()
	defer func() {
		h.mu.Lock()
		delete(h.sessions, s)
		if s.imvuID != 0 {
			h.notifyLocked()
		}
		h.mu.Unlock()
		conn.CloseNow()
	}()

	ctx := r.Context()
	send := func(m message) error {
		b, _ := json.Marshal(m)
		wctx, cancel := context.WithTimeout(ctx, 10*time.Second)
		defer cancel()
		return conn.Write(wctx, websocket.MessageText, b)
	}
	if send(message{Type: "welcome"}) != nil {
		return
	}
	// Keep the connection alive through proxies and NATs.
	go func() {
		t := time.NewTicker(30 * time.Second)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				pctx, cancel := context.WithTimeout(ctx, 10*time.Second)
				err := conn.Ping(pctx)
				cancel()
				if err != nil {
					conn.CloseNow()
					return
				}
			}
		}
	}()
	for {
		_, data, err := conn.Read(ctx)
		if err != nil {
			return
		}
		var m message
		if json.Unmarshal(data, &m) != nil {
			send(message{Type: "error", Message: "not a message"})
			continue
		}
		switch m.Type {
		case "session":
			owner, ok, err := h.owners.LinkOwner(ctx, m.IMVUID)
			if err != nil {
				send(message{Type: "error", Message: "try again later"})
				continue
			}
			if !ok || owner != personID {
				// Not this person's account: no presence, no outfit for it.
				h.set(s, func() { s.imvuID, s.outfit = 0, nil })
				send(message{Type: "error", Message: "not_linked"})
				continue
			}
			h.set(s, func() { s.imvuID, s.since, s.outfit = m.IMVUID, time.Now(), nil })
		case "outfit":
			if len(m.Items) > 64 {
				send(message{Type: "error", Message: "too many items"})
				continue
			}
			items := make([]Item, 0, len(m.Items))
			for _, it := range m.Items {
				if it.ID == "" || len(it.ID) > 128 || len(it.Name) > 128 {
					continue
				}
				items = append(items, it)
			}
			h.set(s, func() { s.outfit = items })
		default:
			send(message{Type: "error", Message: "unknown message type"})
		}
	}
}

func (h *Hub) set(s *session, change func()) {
	h.mu.Lock()
	defer h.mu.Unlock()
	change()
	h.notifyLocked()
}

// ErrUnauthorized is returned by Authorize funcs for a missing or bad token.
var ErrUnauthorized = errors.New("unauthorized")