CosmicClash/server/api/events.go

package api

import (
	"bufio"
	"crypto/sha1"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/cosmic-clash/cosmic-clash/server/domain"
)

const (
	webSocketGUID                = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
	maxWebSocketFrame            = 64 << 10
	eventQueueCapacity           = 32
	webSocketIdleLimit           = 2 * time.Minute
	webSocketWriteLimit          = 10 * time.Second
	webSocketMessageLimit        = 120
	webSocketMessageWindow       = time.Minute
	maxEventConnectionsPerPlayer = 2
	controlPlaneResourceIDPattern = `^[A-Za-z0-9_-]{16,128}$`
)

var controlPlaneResourceIDRE = regexp.MustCompile(controlPlaneResourceIDPattern)

// ControlPlaneEvent is the server-to-client envelope defined by the v1
// WebSocket contract. PlayerID is routing metadata and is never serialized.
type ControlPlaneEvent struct {
	Event      string    `json:"event"`
	Revision   uint64    `json:"revision"`
	ResourceID string    `json:"resource_id"`
	OccurredAt time.Time `json:"occurred_at"`
	State      string    `json:"state,omitempty"`
	Code       string    `json:"code,omitempty"`
	MatchID    string    `json:"match_id,omitempty"`
	ServerID   string    `json:"server_id,omitempty"`
	PlayerID   string    `json:"-"`
}

type eventSubscriber struct {
	playerID string
	queue    chan []byte
}

type eventHub struct {
	mu          sync.Mutex
	subscribers map[*eventSubscriber]struct{}
}

func newEventHub() *eventHub {
	return &eventHub{subscribers: make(map[*eventSubscriber]struct{})}
}

func (h *eventHub) subscribe(playerID string) *eventSubscriber {
	h.mu.Lock()
	connections := 0
	for subscriber := range h.subscribers {
		if subscriber.playerID == playerID {
			connections++
		}
	}
	if connections >= maxEventConnectionsPerPlayer {
		h.mu.Unlock()
		return nil
	}
	subscriber := &eventSubscriber{playerID: playerID, queue: make(chan []byte, eventQueueCapacity)}
	h.subscribers[subscriber] = struct{}{}
	h.mu.Unlock()
	return subscriber
}

func (h *eventHub) unsubscribe(subscriber *eventSubscriber) {
	h.mu.Lock()
	if _, subscribed := h.subscribers[subscriber]; !subscribed {
		h.mu.Unlock()
		return
	}
	delete(h.subscribers, subscriber)
	close(subscriber.queue)
	h.mu.Unlock()
}

func (h *eventHub) publish(event ControlPlaneEvent) error {
	if err := validateControlPlaneEvent(event); err != nil {
		return err
	}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	for subscriber := range h.subscribers {
		if subscriber.playerID != event.PlayerID {
			continue
		}
		select {
		case subscriber.queue <- payload:
		default:
			// A slow client must not block state publication for other clients.
			// Closing its queue makes the connection fail closed and recover via
			// REST resync rather than silently dropping an unbounded history.
			delete(h.subscribers, subscriber)
			close(subscriber.queue)
		}
	}
	return nil
}

func validateControlPlaneEvent(event ControlPlaneEvent) error {
	if event.PlayerID == "" || !controlPlaneResourceIDRE.MatchString(event.ResourceID) || event.OccurredAt.IsZero() {
		return errors.New("invalid control-plane event envelope")
	}
	switch event.Event {
	case "state_changed":
		if !eventState(event.State, "QUEUED", "PROPOSED", "ACCEPTED", "ALLOCATING", "PROCESS_READY", "ASSIGNMENT_READY", "ASSIGNED", "CONNECTING", "LIVE", "RESULT_PENDING", "COMPLETED", "CANCELLED", "EXPIRED", "FAILED") {
			return errors.New("invalid state-changed event")
		}
	case "proposal_changed":
		if !eventState(event.State, "OPEN", "ACCEPTED", "DECLINED", "EXPIRED", "CANCELLED") {
			return errors.New("invalid proposal-changed event")
		}
	case "assignment_changed":
		if !controlPlaneResourceIDRE.MatchString(event.MatchID) || !controlPlaneResourceIDRE.MatchString(event.ServerID) {
			return errors.New("invalid assignment-changed event")
		}
	case "error":
		if !eventState(event.Code, "REVISION_GAP", "NOT_AUTHORISED", "INVALID_STATE", "RATE_LIMITED") {
			return errors.New("invalid error event")
		}
	default:
		return errors.New("unknown control-plane event")
	}
	return nil
}

func eventState(value string, allowed ...string) bool {
	for _, candidate := range allowed {
		if value == candidate {
			return true
		}
	}
	return false
}

func (s *Service) controlPlaneEvent(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
		return
	}
	if !isWebSocketUpgrade(r) || r.Header.Get("Sec-WebSocket-Version") != "13" || !validWebSocketKey(r.Header.Get("Sec-WebSocket-Key")) {
		writeError(w, http.StatusBadRequest, "invalid_websocket_upgrade")
		return
	}
	playerID, ok := s.authenticate(w, r)
	if !ok {
		return
	}
	hub := s.getEventHub()
	subscriber := hub.subscribe(playerID)
	if subscriber == nil {
		writeError(w, http.StatusTooManyRequests, "websocket_connection_limited")
		return
	}
	defer hub.unsubscribe(subscriber)
	hijacker, ok := w.(http.Hijacker)
	if !ok {
		writeError(w, http.StatusNotImplemented, "websocket_unavailable")
		return
	}
	connection, buffered, err := hijacker.Hijack()
	if err != nil {
		return
	}
	defer connection.Close()
	accept := websocketAccept(r.Header.Get("Sec-WebSocket-Key"))
	if _, err := fmt.Fprintf(buffered, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: %s\r\n\r\n", accept); err != nil {
		return
	}
	if err := buffered.Flush(); err != nil {
		return
	}
	var writeMu sync.Mutex
	done := make(chan struct{})
	go func() {
		defer close(done)
		readWebSocketFrames(connection, &writeMu)
	}()
	for {
		select {
		case payload, open := <-subscriber.queue:
			if !open {
				return
			}
			writeMu.Lock()
			err := writeWebSocketFrameWithDeadline(connection, 0x1, payload, webSocketWriteLimit)
			writeMu.Unlock()
			if err != nil {
				return
			}
		case <-done:
			return
		}
	}
}

func (s *Service) getEventHub() *eventHub {
	s.eventsMu.Lock()
	defer s.eventsMu.Unlock()
	if s.events == nil {
		s.events = newEventHub()
	}
	return s.events
}

// PublishControlPlaneEvent routes an already-authorized event to the matching
// authenticated player connection on THIS replica. Durable callers should
// publish from their outbox after commit; this in-memory hub is deliberately
// non-authoritative.
func (s *Service) PublishControlPlaneEvent(event ControlPlaneEvent) error {
	return s.getEventHub().publish(event)
}

// fannedOutEvent is the fan-out wire shape. It cannot reuse ControlPlaneEvent
// directly because that type hides PlayerID from clients (json:"-"), and the
// recipient is precisely what a peer replica needs in order to route.
type fannedOutEvent struct {
	ControlPlaneEvent
	PlayerID string `json:"player_id"`
}

// EncodeFannedOutEvent and DecodeFannedOutEvent are exported for the
// control-plane binary, which owns the transport wiring.
func EncodeFannedOutEvent(event ControlPlaneEvent) ([]byte, error) {
	return json.Marshal(fannedOutEvent{ControlPlaneEvent: event, PlayerID: event.PlayerID})
}

func DecodeFannedOutEvent(payload []byte) (ControlPlaneEvent, error) {
	var decoded fannedOutEvent
	if err := json.Unmarshal(payload, &decoded); err != nil {
		return ControlPlaneEvent{}, err
	}
	event := decoded.ControlPlaneEvent
	event.PlayerID = decoded.PlayerID
	if event.Event == "" || event.ResourceID == "" || event.PlayerID == "" {
		return ControlPlaneEvent{}, fmt.Errorf("invalid fanned-out control-plane event")
	}
	return event, nil
}

// publishOutboxEvent is how the outbox dispatchers publish. When EventFanout
// is configured it hands the event to the shared transport so every replica --
// including whichever one holds the subscriber's WebSocket -- can deliver it.
// Without it, behaviour is unchanged: local-hub only, correct for a single
// replica and for tests.
func (s *Service) publishOutboxEvent(event ControlPlaneEvent) error {
	if s.EventFanout != nil {
		return s.EventFanout(event)
	}
	return s.PublishControlPlaneEvent(event)
}

func (s *Service) publishTicketEvent(ticket domain.QueueTicket, now time.Time) {
	_ = s.PublishControlPlaneEvent(ControlPlaneEvent{
		Event: "state_changed", Revision: ticket.Revision, ResourceID: ticket.TicketID,
		OccurredAt: now, State: string(ticket.State), PlayerID: ticket.PlayerID,
	})
}

func (s *Service) publishProposalEvent(proposal domain.Proposal, now time.Time) {
	for _, participant := range proposal.Participants {
		_ = s.PublishControlPlaneEvent(ControlPlaneEvent{
			Event: "proposal_changed", Revision: proposal.Revision, ResourceID: proposal.ProposalID,
			OccurredAt: now, State: string(proposal.State), PlayerID: participant.PlayerID,
		})
	}
}

func isWebSocketUpgrade(r *http.Request) bool {
	return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") && headerContainsToken(r.Header.Values("Connection"), "upgrade")
}

func headerContainsToken(values []string, wanted string) bool {
	for _, value := range values {
		for _, token := range strings.Split(value, ",") {
			if strings.EqualFold(strings.TrimSpace(token), wanted) {
				return true
			}
		}
	}
	return false
}

func websocketAccept(key string) string {
	digest := sha1.Sum([]byte(key + webSocketGUID))
	return base64.StdEncoding.EncodeToString(digest[:])
}

func validWebSocketKey(key string) bool {
	decoded, err := base64.StdEncoding.DecodeString(key)
	return err == nil && len(decoded) == 16
}

func readWebSocketFrames(connection net.Conn, writeMu *sync.Mutex) {
	reader := bufio.NewReader(connection)
	windowStarted := time.Now()
	messageCount := 0
	for {
		if err := connection.SetReadDeadline(time.Now().Add(webSocketIdleLimit)); err != nil {
			return
		}
		opcode, _, err := readWebSocketFrame(reader)
		if err != nil || opcode == 0x8 {
			return
		}
		now := time.Now()
		if !allowWebSocketMessage(now, &windowStarted, &messageCount) {
			return
		}
		if opcode == 0x9 {
			writeMu.Lock()
			_ = writeWebSocketFrameWithDeadline(connection, 0xA, nil, webSocketWriteLimit)
			writeMu.Unlock()
		}
	}
}

func allowWebSocketMessage(now time.Time, windowStarted *time.Time, count *int) bool {
	if windowStarted == nil || count == nil || now.IsZero() {
		return false
	}
	if !now.Before(windowStarted.Add(webSocketMessageWindow)) {
		*windowStarted = now
		*count = 0
	}
	if *count >= webSocketMessageLimit {
		return false
	}
	*count++
	return true
}

func readWebSocketFrame(reader *bufio.Reader) (byte, []byte, error) {
	first, err := reader.ReadByte()
	if err != nil {
		return 0, nil, err
	}
	second, err := reader.ReadByte()
	if err != nil {
		return 0, nil, err
	}
	if first&0x70 != 0 || first&0x80 == 0 {
		return 0, nil, errors.New("unsupported websocket frame")
	}
	opcode := first & 0x0f
	if opcode != 0x8 && opcode != 0x9 && opcode != 0xA {
		return 0, nil, errors.New("unsupported websocket opcode")
	}
	if second&0x80 == 0 {
		return 0, nil, errors.New("unmasked websocket frame")
	}
	length := int64(second & 0x7f)
	if length == 126 {
		var extended uint16
		if err := binary.Read(reader, binary.BigEndian, &extended); err != nil {
			return 0, nil, err
		}
		length = int64(extended)
	} else if length == 127 {
		var extended uint64
		if err := binary.Read(reader, binary.BigEndian, &extended); err != nil || extended > maxWebSocketFrame {
			return 0, nil, errors.New("websocket frame too large")
		}
		length = int64(extended)
	}
	if length > maxWebSocketFrame {
		return 0, nil, errors.New("websocket frame too large")
	}
	if opcode&0x8 != 0 && length > 125 {
		return 0, nil, errors.New("websocket control frame too large")
	}
	var mask [4]byte
	if _, err := io.ReadFull(reader, mask[:]); err != nil {
		return 0, nil, err
	}
	payload := make([]byte, length)
	if _, err := io.ReadFull(reader, payload); err != nil {
		return 0, nil, err
	}
	for i := range payload {
		payload[i] ^= mask[i%4]
	}
	return opcode, payload, nil
}

func writeWebSocketFrame(connection net.Conn, opcode byte, payload []byte) error {
	if len(payload) > maxWebSocketFrame {
		return errors.New("websocket frame too large")
	}
	header := []byte{0x80 | opcode}
	switch {
	case len(payload) < 126:
		header = append(header, byte(len(payload)))
	case len(payload) <= 65535:
		header = append(header, 126, 0, 0)
		binary.BigEndian.PutUint16(header[len(header)-2:], uint16(len(payload)))
	default:
		header = append(header, 127)
		var extended [8]byte
		binary.BigEndian.PutUint64(extended[:], uint64(len(payload)))
		header = append(header, extended[:]...)
	}
	if _, err := connection.Write(header); err != nil {
		return err
	}
	_, err := connection.Write(payload)
	return err
}

func writeWebSocketFrameWithDeadline(connection net.Conn, opcode byte, payload []byte, timeout time.Duration) error {
	if timeout <= 0 {
		return errors.New("invalid websocket write timeout")
	}
	if err := connection.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
		return err
	}
	return writeWebSocketFrame(connection, opcode, payload)
}