package gateway import ( "encoding/json" "log/slog" "sync" ) // Hub maintains the set of active clients and broadcasts messages to them. type Hub struct { mu sync.RWMutex clients map[*Client]struct{} register chan *Client unregister chan *Client broadcast chan []byte logger *slog.Logger } // NewHub creates a new Hub. func NewHub(logger *slog.Logger) *Hub { return &Hub{ clients: make(map[*Client]struct{}), register: make(chan *Client), unregister: make(chan *Client), broadcast: make(chan []byte, 256), logger: logger, } } // Run starts the hub's main event loop. Call this in a goroutine. func (h *Hub) Run() { for { select { case client := <-h.register: h.mu.Lock() h.clients[client] = struct{}{} h.mu.Unlock() h.logger.Info("client connected", "user_id", client.UserID) case client := <-h.unregister: h.mu.Lock() if _, ok := h.clients[client]; ok { delete(h.clients, client) close(client.send) } h.mu.Unlock() h.logger.Info("client disconnected", "user_id", client.UserID) case message := <-h.broadcast: h.mu.RLock() for client := range h.clients { select { case client.send <- message: default: go func(c *Client) { h.unregister <- c }(client) } } h.mu.RUnlock() } } } // Register queues a client for registration with the hub. func (h *Hub) Register(client *Client) { h.register <- client } // Unregister queues a client for removal from the hub. func (h *Hub) Unregister(client *Client) { h.unregister <- client } // BroadcastEvent serializes an Event and broadcasts it to all connected clients. func (h *Hub) BroadcastEvent(event Event) { data, err := json.Marshal(event) if err != nil { h.logger.Error("failed to marshal event", "type", event.Type, "error", err) return } h.broadcast <- data } // ClientCount returns the number of connected clients. func (h *Hub) ClientCount() int { h.mu.RLock() defer h.mu.RUnlock() return len(h.clients) }