casino/cmd/arcade/hub.go

package main

import (
	"context"
	"errors"
	"log"
	"net"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/drjones/quantum-arcade/pkg/cluster"
	"github.com/drjones/quantum-arcade/pkg/room"
)

// gameHub owns one game across the cluster.
//
// Exactly one instance leads a game: it runs the round loop, drives the
// simulation, settles to the ledger, and publishes each frame. Every other
// instance relays those frames to its own connected clients. Clients cannot
// tell the difference, and neither can the ledger.
//
// Roles are renegotiated on a timer rather than agreed once, so an instance
// disappearing is not a special case — its lease simply stops being renewed
// and the next campaign hands the game to someone else.
type gameHub struct {
	game string
	room *room.Room
	node *cluster.Node

	mu      sync.RWMutex
	leading bool
	subs    map[chan []byte]struct{}

	// lastFrame is the most recent frame this instance saw, whether it
	// produced it or relayed it. A follower's own room object sits idle, so
	// this — not the local room — is what any read of "current state" must
	// use, or a follower would report a permanently settled game.
	lastFrame []byte

	// cancelLead stops the leader's round loop when leadership is lost.
	cancelLead context.CancelFunc
}

func newGameHub(game string, r *room.Room, node *cluster.Node) *gameHub {
	return &gameHub{
		game: game,
		room: r,
		node: node,
		subs: make(map[chan []byte]struct{}),
	}
}

// Subscribe returns frames for this game, whether this instance is producing
// them or relaying them.
func (h *gameHub) Subscribe() (<-chan []byte, func()) {
	ch := make(chan []byte, 4)
	h.mu.Lock()
	h.subs[ch] = struct{}{}
	h.mu.Unlock()
	return ch, func() {
		h.mu.Lock()
		delete(h.subs, ch)
		close(ch)
		h.mu.Unlock()
	}
}

// fanout delivers a frame to this instance's own clients. A client that has
// stopped reading is skipped rather than allowed to stall the game.
func (h *gameHub) fanout(payload []byte) {
	h.mu.Lock()
	h.lastFrame = payload
	h.mu.Unlock()

	h.mu.RLock()
	defer h.mu.RUnlock()
	for ch := range h.subs {
		select {
		case ch <- payload:
		default:
		}
	}
}

// LastFrame returns the most recent state this instance knows about, and
// whether it has seen one yet.
func (h *gameHub) LastFrame() ([]byte, bool) {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.lastFrame, len(h.lastFrame) > 0
}

// Leading reports whether this instance currently drives the game.
func (h *gameHub) Leading() bool {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.leading
}

// supervise campaigns for the game and switches roles as leadership moves.
func (h *gameHub) supervise(ctx context.Context) {
	// Followers hold a subscription to the cluster's frame channel. It is torn
	// down on promotion so a leader never relays its own frames back to itself.
	var stopRelay func()
	defer func() {
		if stopRelay != nil {
			stopRelay()
		}
	}()

	ticker := time.NewTicker(cluster.RenewInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			h.demote()
			return
		case <-ticker.C:
		}

		var held bool
		var err error
		if h.Leading() {
			held, err = h.node.Renew(ctx, h.game)
		} else {
			held, err = h.node.Campaign(ctx, h.game)
		}
		if err != nil {
			// Redis is unreachable. A current leader keeps running rather than
			// abandoning a round mid-flight; its lease will expire and another
			// instance will take over if the outage outlasts it.
			log.Printf("hub %s: coordination error: %v", h.game, err)
			continue
		}

		switch {
		case held && !h.Leading():
			if stopRelay != nil {
				stopRelay()
				stopRelay = nil
			}
			h.promote(ctx)

		case !held && h.Leading():
			h.demote()
			stopRelay = h.startRelay(ctx)

		case !held && stopRelay == nil:
			// Follower with no relay yet — subscribe so clients see the game.
			stopRelay = h.startRelay(ctx)
		}
	}
}

// promote starts driving the game on this instance.
func (h *gameHub) promote(ctx context.Context) {
	leadCtx, cancel := context.WithCancel(ctx)

	h.mu.Lock()
	h.leading = true
	h.cancelLead = cancel
	h.mu.Unlock()

	log.Printf("hub %s: leading", h.game)

	// Forward the room's frames both to this instance's clients and to peers.
	frames, unsubscribe := h.room.Subscribe()
	go func() {
		defer unsubscribe()
		for {
			select {
			case <-leadCtx.Done():
				return
			case payload, ok := <-frames:
				if !ok {
					return
				}
				h.fanout(payload)
				if err := h.node.PublishFrame(leadCtx, h.game, payload); err != nil &&
					!errors.Is(err, context.Canceled) {
					log.Printf("hub %s: publishing frame: %v", h.game, err)
				}
			}
		}
	}()

	go func() {
		if err := h.room.Run(leadCtx); err != nil && !errors.Is(err, context.Canceled) {
			log.Printf("hub %s: round loop stopped: %v", h.game, err)
		}
	}()
}

// demote stops driving the game.
func (h *gameHub) demote() {
	h.mu.Lock()
	wasLeading := h.leading
	h.leading = false
	cancel := h.cancelLead
	h.cancelLead = nil
	h.mu.Unlock()

	if cancel != nil {
		cancel()
	}
	if wasLeading {
		log.Printf("hub %s: no longer leading", h.game)
	}
}

// startRelay subscribes to the leader's frames and passes them to this
// instance's clients.
func (h *gameHub) startRelay(ctx context.Context) func() {
	frames, unsubscribe := h.node.SubscribeFrames(ctx, h.game)
	relayCtx, cancel := context.WithCancel(ctx)

	go func() {
		for {
			select {
			case <-relayCtx.Done():
				return
			case payload, ok := <-frames:
				if !ok {
					return
				}
				h.fanout(payload)
			}
		}
	}()

	return func() {
		cancel()
		unsubscribe()
	}
}

// advertiseAddr is how peers reach this instance.
//
// It prefers an explicit setting, then the first non-loopback address it can
// find — so a cloned VM that gets its address from DHCP advertises correctly
// without being told what it is.
func advertiseAddr() string {
	if v := os.Getenv("ARCADE_ADVERTISE"); v != "" {
		return v
	}
	port := os.Getenv("ARCADE_ADDR")
	if port == "" {
		port = ":8080"
	}
	if !strings.HasPrefix(port, ":") {
		if _, p, err := net.SplitHostPort(port); err == nil {
			port = ":" + p
		}
	}

	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return "127.0.0.1" + port
	}
	for _, a := range addrs {
		ipnet, ok := a.(*net.IPNet)
		if !ok || ipnet.IP.IsLoopback() || ipnet.IP.To4() == nil {
			continue
		}
		return ipnet.IP.String() + port
	}
	return "127.0.0.1" + port
}