pulse-lets-go/internal/api/ws.go

package api

import (
	"fmt"
	"log"
	"net/http"
	"sync"

	"github.com/golang-jwt/jwt/v5"
	"github.com/gorilla/websocket"

	"github.com/pulse-lets-go/internal/models"
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true // Разрешаем все origins для dev
	},
}

const wsSendBufferSize = 32 // размер буфера per-client канала

// wsClient — соединение WebSocket одного клиента с выделенной write-горутиной.
type wsClient struct {
	conn *websocket.Conn
	send chan *models.WsMetricsMessage
}

// wsHub управляет подключёнными WebSocket-клиентами и рассылает метрики.
type wsHub struct {
	mu      sync.RWMutex
	clients map[*wsClient]bool
}

// newWSHub создаёт хаб WebSocket.
func newWSHub() *wsHub {
	return &wsHub{
		clients: make(map[*wsClient]bool),
	}
}

// add добавляет клиента в хаб.
func (h *wsHub) add(c *wsClient) {
	h.mu.Lock()
	h.clients[c] = true
	h.mu.Unlock()
}

// remove удаляет клиента из хаба и закрывает его канал.
// Идемпотентно — может вызываться многократно (writePump + readPump).
func (h *wsHub) remove(c *wsClient) {
	h.mu.Lock()
	if h.clients[c] {
		delete(h.clients, c)
		close(c.send)
		c.conn.Close()
	}
	h.mu.Unlock()
}

// broadcast рассылает метрики всем клиентам неблокирующе.
// Медленные клиенты пропускают сообщения (best-effort delivery).
func (h *wsHub) broadcast(msg *models.WsMetricsMessage) {
	h.mu.RLock()
	defer h.mu.RUnlock()

	for client := range h.clients {
		select {
		case client.send <- msg:
		default:
			// Клиент слишком медленный — пропускаем сообщение
			go h.remove(client)
		}
	}
}

// writePump — выделенная горутина записи для одного клиента.
// Читает из client.send и пишет в WebSocket.
func (c *wsClient) writePump(hub *wsHub) {
	defer hub.remove(c)
	for msg := range c.send {
		if err := c.conn.WriteJSON(msg); err != nil {
			return
		}
	}
}

// handleWS — WebSocket /ws/metrics.
func (a *API) handleWS(w http.ResponseWriter, r *http.Request) {
	token := r.URL.Query().Get("token")
	if token == "" {
		// пробуем заголовок
		token = extractBearerToken(r)
	}
	if token == "" {
		writeError(w, http.StatusUnauthorized, "требуется токен")
		return
	}

	// Валидация JWT
	userInfo, err := a.validateToken(token)
	if err != nil {
		writeError(w, http.StatusUnauthorized, "недействительный токен")
		return
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("[ws] ошибка upgrade: %v", err)
		return
	}

	client := &wsClient{
		conn: conn,
		send: make(chan *models.WsMetricsMessage, wsSendBufferSize),
	}
	a.wsHub.add(client)

	log.Printf("[ws] клиент подключён: %s (%s)", userInfo.Username, r.RemoteAddr)

	// Запускаем write-горутину
	go client.writePump(a.wsHub)

	// Отправляем текущее состояние при подключении
	go func() {
		a.BroadcastAll()
	}()

	// Читаем из вебсокета (ping/pong + закрытие)
	go func() {
		defer a.wsHub.remove(client)
		for {
			_, _, err := conn.ReadMessage()
			if err != nil {
				log.Printf("[ws] клиент отключён: %s", r.RemoteAddr)
				return
			}
		}
	}()
}

// BroadcastMetrics рассылает метрики нод всем WS-клиентам.
func (a *API) BroadcastMetrics() {
	nodes := a.engine.GetAllNodes()
	msg := &models.WsMetricsMessage{
		Type:    "nodes_update",
		Payload: nodes,
	}
	a.wsHub.broadcast(msg)
}

// BroadcastAll рассылает ноды и состояние балансировщика.
func (a *API) BroadcastAll() {
	a.BroadcastMetrics()
	a.broadcastBalancerUpdate()
}

func (a *API) broadcastBalancerUpdate() {
	stats := a.engine.GetHealthStats()
	gwTotal, gwUp, gwDown := a.gatewayStats()

	info := models.BalancerInfo{
		NatsConnected:  a.natsConnected != nil && a.natsConnected(),
		GatewaysTotal:  gwTotal,
		GatewaysUp:     gwUp,
		GatewaysDown:   gwDown,
		RouteTotal:     stats.RouteRequests,
		RouteFallbacks: stats.RouteFallbacks,
		UptimeSec:      stats.UptimeSeconds,
		BalanceMode:    a.engine.GetBalancingMode(),
	}

	if a.eslClient != nil {
		eslStats := a.eslClient.GetStats()
		info.EslConnected = eslStats.Status == "connected"
		info.EslUptimeSec = eslStats.UptimeSec
		info.EslReconnects = eslStats.Reconnects

		a.fsStatsMu.RLock()
		info.FsActiveCalls = a.fsStats.ActiveCalls
		info.FsMaxSessions = a.fsStats.MaxSessions
		info.FsUptime = a.fsStats.FsUptime
		a.fsStatsMu.RUnlock()
	}

	msg := &models.WsMetricsMessage{
		Type:    "balancer_update",
		Payload: info,
	}
	a.wsHub.broadcast(msg)
}

// validateToken проверяет JWT и возвращает UserInfo.
func (a *API) validateToken(tokenStr string) (*UserInfo, error) {
	token, err := parseJWT(tokenStr, a.jwtSecret)
	if err != nil {
		return nil, err
	}
	return token, nil
}

// parseJWT парсит JWT токен.
func parseJWT(tokenStr string, secret string) (*UserInfo, error) {
	c := &claims{}
	token, err := jwt.ParseWithClaims(tokenStr, c, func(token *jwt.Token) (interface{}, error) {
		return []byte(secret), nil
	})
	if err != nil || !token.Valid {
		return nil, fmt.Errorf("недействительный токен: %w", err)
	}
	return &UserInfo{
		UserID:   c.UserID,
		Username: c.Username,
		Role:     c.Role,
	}, nil
}