feat: WebSocket heartbeat + pub/sub

ws_client.go (new): Client struct with readPump/writePump goroutines, send channel, ping/pong heartbeat (60s pong, 54s ping interval)

websocket.go (rewrite): Hub with register/unregister/broadcast channels, Run() goroutine, typed events ({type, data} envelope)

main.go: go wsHub.Run() goroutine

Breaking: wire format changed from flat VisitorDot to typed messages

FINDINGS.md

@@ -16,10 +16,11 @@
 6. **JWT секреты без дефолтов в production** — config.go требует JWT_SECRET, JWT_REFRESH_SECRET, ALLOWED_ORIGINS в production; дефолты ("dev-secret", "localhost:3000") запрещены, приложение паникует при старте.
 7. **Token Reuse Detection** — уже реализован: `GetRevoked` в репозитории, проверка в `RefreshSession`, отзыв всех токенов пользователя при обнаружении, обработка в хендлере.
 8. **Скрытие внутренних ошибок в production** — все хендлеры используют единый `writeError` с параметром `err` для логирования; в production 500 ошибки возвращают "internal server error" вместо деталей БД/логики.
+9. **WebSocket heartbeat + pub/sub** — рефакторинг: Hub с `register`/`unregister`/`broadcast` каналами и `Run()` горутиной; `Client` с `send chan`, `readPump`/`writePump` горутинами; ping/pong heartbeat (60s pong wait, 54s ping interval); typed events (`{"type":"visitors","data":{...}}`) вместо плоского JSON. Breaking change: старый формат `{"visitors":[...]}` больше не поддерживается.
 
 ### ✅ P1 — Исправлено
-9. **Теги и фичи в листинге мест** — `PlaceFilter.IncludeTagsFeatures` + batch-загрузка (`GetTagsBatch`/`GetFeaturesBatch`), запрос `?include=tags,features`. N+1 проблема решена.
-10. **Cursor-based пагинация** — `PlaceFilter.CursorRating`, `PaginatedPlaces` с json-тегами, сервис возвращает `*PaginatedPlaces`, репозиторий поддерживает курсор для `created_at` и `rating` сортировок. API: `?cursor=<base64>&limit=N`. Ответ: `{"data":[...], "next_cursor":"...", "has_more":true}`.
+10. **Теги и фичи в листинге мест** — `PlaceFilter.IncludeTagsFeatures` + batch-загрузка (`GetTagsBatch`/`GetFeaturesBatch`), запрос `?include=tags,features`. N+1 проблема решена.
+11. **Cursor-based пагинация** — `PlaceFilter.CursorRating`, `PaginatedPlaces` с json-тегами, сервис возвращает `*PaginatedPlaces`, репозиторий поддерживает курсор для `created_at` и `rating` сортировок. API: `?cursor=<base64>&limit=N`. Ответ: `{"data":[...], "next_cursor":"...", "has_more":true}`.
 
 ### ✅ P1 — Исправлено
 6. **Rate limiter** — Redis-based имплементация подключена в main.go, in-memory как fallback при недоступности Redis (реализован корректный fallback: при недоступности Redis автоматически используется in-memory limiter; в production `failOpen=false` — возвращается 503 только если оба лимитера недоступны).
@@ -66,6 +67,8 @@
 | `backend/cmd/api/main.go` | isProd для WS Hub; rate limiter fallback (Redis → in-memory) |
 | `backend/internal/config/config.go` | JWT секреты обязательны в production, запрещены дефолты |
 | `backend/internal/handlers/errors.go` | **Новый**: единый writeError с логированием и скрытием деталей в prod |
+| `backend/internal/handlers/ws_client.go` | **Новый**: Client readPump/writePump, ping/pong heartbeat |
+| `backend/internal/handlers/websocket.go` | **Переписан**: Hub с register/unregister/broadcast, Run(), typed events |
 | `backend/internal/handlers/*.go` | Все хендлеры обновлены: writeError(msg, err) вместо err.Error() |
 | `backend/internal/models/place.go` | PlaceFilter.IncludeTagsFeatures, CursorRating; PaginatedPlaces json-теги |
 | `backend/internal/services/places.go` | List: batch-загрузка тегов/фич, encodeCursor/DecodeCursor, возвращает *PaginatedPlaces |

backend/cmd/api/main.go

@@ -76,6 +76,7 @@ func main() {
 	bookingHandler := handlers.NewBookingHandler(bookingRepo, placeRepo, cfg.AppEnv)
 	tagHandler := handlers.NewTagHandler(tagRepo, featureRepo, cfg.AppEnv)
 	wsHub := handlers.NewWSHub(cfg.AllowedOrigins, cfg.AppEnv == "production")
+	go wsHub.Run()
 	uploadHandler, err := handlers.NewUploadHandler(cfg.S3Endpoint, cfg.S3PublicEndpoint, cfg.S3AccessKey, cfg.S3SecretKey, cfg.S3Bucket, false, cfg.AppEnv)
 	if err != nil {
 		logger.ErrorContext(ctx, "upload handler init failed", log.WithError(err))

backend/internal/handlers/websocket.go

@@ -1,10 +1,9 @@
 package handlers
 
 import (
+	"encoding/json"
 	"log"
 	"net/http"
-	"sync"
-	"time"
 
 	"github.com/gorilla/websocket"
 )
@@ -15,24 +14,86 @@ type VisitorDot struct {
 	Lng    float64 `json:"lng"`
 }
 
+type visitorsPayload struct {
+	Visitors []VisitorDot `json:"visitors"`
+}
+
 type WSHub struct {
-	mu             sync.RWMutex
-	clients        map[*websocket.Conn]VisitorDot
+	clients        map[*Client]bool
+	register       chan *Client
+	unregister     chan *Client
+	broadcast      chan []byte
 	allowedOrigins []string
 	isProd         bool
 }
 
 func NewWSHub(allowedOrigins []string, isProd bool) *WSHub {
-	if isProd && len(allowedOrigins) == 0 {
-		log.Println("WARN: WebSocket AllowedOrigins is empty in production — connections will be rejected")
-	}
 	return &WSHub{
-		clients:        make(map[*websocket.Conn]VisitorDot),
+		clients:        make(map[*Client]bool),
+		register:       make(chan *Client),
+		unregister:     make(chan *Client),
+		broadcast:      make(chan []byte, 256),
 		allowedOrigins: allowedOrigins,
 		isProd:         isProd,
 	}
 }
 
+func (h *WSHub) Run() {
+	for {
+		select {
+		case client := <-h.register:
+			h.clients[client] = true
+			h.broadcastVisitors()
+
+		case client := <-h.unregister:
+			if _, ok := h.clients[client]; ok {
+				delete(h.clients, client)
+				close(client.send)
+				h.broadcastVisitors()
+			}
+
+		case message := <-h.broadcast:
+			for client := range h.clients {
+				select {
+				case client.send <- message:
+				default:
+					close(client.send)
+					delete(h.clients, client)
+				}
+			}
+		}
+	}
+}
+
+func (h *WSHub) broadcastVisitors() {
+	visitors := make([]VisitorDot, 0, len(h.clients))
+	for c := range h.clients {
+		visitors = append(visitors, VisitorDot{
+			UserID: c.userID,
+			Lat:    c.lat,
+			Lng:    c.lng,
+		})
+	}
+
+	data, err := json.Marshal(visitorsPayload{Visitors: visitors})
+	if err != nil {
+		return
+	}
+	msg, err := json.Marshal(wsMessage{Type: "visitors", Data: data})
+	if err != nil {
+		return
+	}
+
+	for client := range h.clients {
+		select {
+		case client.send <- msg:
+		default:
+			close(client.send)
+			delete(h.clients, client)
+		}
+	}
+}
+
 func (h *WSHub) HandleWS(w http.ResponseWriter, r *http.Request) {
 	upgrader := websocket.Upgrader{
 		CheckOrigin: func(r *http.Request) bool {
@@ -55,45 +116,14 @@ func (h *WSHub) HandleWS(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	dot := VisitorDot{}
-
-	h.mu.Lock()
-	h.clients[conn] = dot
-	h.mu.Unlock()
-
-	for {
-		var msg VisitorDot
-		if err := conn.ReadJSON(&msg); err != nil {
-			break
-		}
-
-		h.mu.Lock()
-		h.clients[conn] = msg
-		visitors := make([]VisitorDot, 0, len(h.clients))
-		for _, v := range h.clients {
-			visitors = append(visitors, v)
-		}
-
-		var failed []*websocket.Conn
-		for c := range h.clients {
-			if err := c.SetWriteDeadline(time.Now().Add(10 * time.Second)); err != nil {
-				failed = append(failed, c)
-				continue
-			}
-			if err := c.WriteJSON(map[string]interface{}{"visitors": visitors}); err != nil {
-				failed = append(failed, c)
-			}
-		}
-		for _, c := range failed {
-			c.WriteMessage(websocket.CloseMessage, []byte{})
-			c.Close()
-			delete(h.clients, c)
-		}
-		h.mu.Unlock()
+	client := &Client{
+		hub:  h,
+		conn: conn,
+		send: make(chan []byte, 256),
 	}
 
-	h.mu.Lock()
-	delete(h.clients, conn)
-	h.mu.Unlock()
-	conn.Close()
+	h.register <- client
+
+	go client.writePump()
+	go client.readPump()
 }

backend/internal/handlers/ws_client.go

@@ -0,0 +1,89 @@
+package handlers
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+const (
+	writeWait      = 10 * time.Second
+	pongWait       = 60 * time.Second
+	pingPeriod     = (pongWait * 9) / 10
+	maxMessageSize = 4096
+)
+
+type wsMessage struct {
+	Type string          `json:"type"`
+	Data json.RawMessage `json:"data"`
+}
+
+type Client struct {
+	hub    *WSHub
+	conn   *websocket.Conn
+	send   chan []byte
+	userID string
+	lat    float64
+	lng    float64
+}
+
+func (c *Client) readPump() {
+	defer func() {
+		c.hub.unregister <- c
+		c.conn.Close()
+	}()
+
+	c.conn.SetReadLimit(maxMessageSize)
+	c.conn.SetReadDeadline(time.Now().Add(pongWait))
+	c.conn.SetPongHandler(func(string) error {
+		c.conn.SetReadDeadline(time.Now().Add(pongWait))
+		return nil
+	})
+
+	for {
+		var msg wsMessage
+		if err := c.conn.ReadJSON(&msg); err != nil {
+			break
+		}
+
+		switch msg.Type {
+		case "visitor_update":
+			var dot VisitorDot
+			if err := json.Unmarshal(msg.Data, &dot); err != nil {
+				continue
+			}
+			c.userID = dot.UserID
+			c.lat = dot.Lat
+			c.lng = dot.Lng
+			c.hub.broadcastVisitors()
+		}
+	}
+}
+
+func (c *Client) writePump() {
+	ticker := time.NewTicker(pingPeriod)
+	defer func() {
+		ticker.Stop()
+		c.conn.Close()
+	}()
+
+	for {
+		select {
+		case message, ok := <-c.send:
+			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if !ok {
+				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
+				return
+			}
+			if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil {
+				return
+			}
+		case <-ticker.C:
+			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				return
+			}
+		}
+	}
+}