package api import ( "bufio" "crypto/sha1" "encoding/base64" "encoding/binary" "encoding/json" "errors" "fmt" "io" "net" "net/http" "strings" "sync" "time" ) 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 event.PlayerID == "" || event.Event == "" || event.ResourceID == "" || event.OccurredAt.IsZero() { return errors.New("invalid control-plane event") } 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 (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 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 }