CosmicClash/server/api/events.go

package api

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

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

const (
	webSocketGUID      = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
	maxWebSocketFrame  = 64 << 10
	eventQueueCapacity = 32
)

// 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 {
	subscriber := &eventSubscriber{playerID: playerID, queue: make(chan []byte, eventQueueCapacity)}
	h.mu.Lock()
	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 == "" || 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 event.MatchID == "" || 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) || !validWebSocketKey(r.Header.Get("Sec-WebSocket-Key")) {
		writeError(w, http.StatusBadRequest, "invalid_websocket_upgrade")
		return
	}
	playerID, ok := s.authenticate(w, r)
	if !ok {
		return
	}
	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
	}
	subscriber := s.getEventHub().subscribe(playerID)
	defer s.getEventHub().unsubscribe(subscriber)

	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 := writeWebSocketFrame(connection, 0x1, payload)
			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. 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)
}

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)
	for {
		opcode, _, err := readWebSocketFrame(reader)
		if err != nil || opcode == 0x8 {
			return
		}
		if opcode == 0x9 {
			writeMu.Lock()
			_ = writeWebSocketFrame(connection, 0xA, nil)
			writeMu.Unlock()
		}
	}
}

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")
	}
	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")
	}
	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 first & 0x0f, 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
}