Seeking guide on ws 10k implementation

[SOG-web]

I have a situation of needing to send 10k ws at a time in my project. I am looking for the best way to implement this as I want it to be performant

I currently have two implementation, please I want suggestions on which will work best or any other suggestion on how to get the result with minimal resource usage

Am using this ws to create realtime update for my db per table and the redis is use to manager clients across all server instance(in case of horizontal scaling with kuberneties)

1. using go routine per table

// Package database implements WebSocket-based real-time database operations
package database


// MessageType represents different types of messages that can be exchanged between client and server
type MessageType string

// Constants defining various types of messages used in WebSocket communication
const (
	MessageTypeSubscribe    MessageType = "subscribe"    // Client requests to subscribe to table updates
	MessageTypeUnsubscribe  MessageType = "unsubscribe"  // Client requests to unsubscribe from table updates
	MessageTypeGetTableData MessageType = "get_table_data" // Client requests data from a specific table
	MessageTypeError        MessageType = "error"        // Server sends error messages to client
	MessageTypeStatus       MessageType = "status"       // Server sends status updates to client
	MessageTypeTableData    MessageType = "table_data"   // Server sends table data to client
	MessageTypeMutation     MessageType = "mutation"     // Server notifies clients about data changes
)

// Rate limiting constants to prevent abuse
const (
	MaxMessagesPerMinute = 100      // Maximum number of messages allowed per client per minute
	RateLimitWindow      = time.Minute // Time window for rate limiting
)

// WebSocketMessage defines the structure for all messages exchanged over WebSocket
type WebSocketMessage struct {
	Type   MessageType  `json:"type"`           // Type of the message
	Tables []string     `json:"tables,omitempty"` // List of tables (used for subscribe/unsubscribe)
	Data   interface{}  `json:"data,omitempty"`  // Payload data
	Error  string       `json:"error,omitempty"` // Error message if any
}

// Client represents a connected WebSocket client and its properties
type Client struct {
	ID     string          // Unique identifier for the client
	Conn   *websocket.Conn // WebSocket connection
	Tables map[string]bool // Map of tables this client is subscribed to
	mu     sync.RWMutex    // Mutex for thread-safe operations
	userId string          // User identifier for authentication
}

// TableSubscribers maintains a map of tables to their subscribed clients
type TableSubscribers struct {
	subscribers map[string]map[*Client]bool
	mu          sync.RWMutex
}

// ConnectionManager coordinates all WebSocket connections and message broadcasting
type ConnectionManager struct {
	clients         map[*Client]bool
	mu             sync.RWMutex
	register       chan *Client
	unregister     chan *Client
	redis          *redisService.RedisService
	ctx            context.Context
	tableSubscribers *TableSubscribers
}

// BroadcastMessage represents a message to be broadcast to subscribed clients
type BroadcastMessage struct {
	Table string          `json:"table"` // Table that was modified
	Data  json.RawMessage `json:"data"`  // Modified data
}

var (
	manager *ConnectionManager
)

// init initializes the ConnectionManager and starts necessary background processes
func init() {
	// Initialize Redis service
	redisInstance, err := redisService.GetInstance()
	if err != nil {
		log.Printf("Failed to initialize Redis service: %v", err)
		return
	}

	// Initialize ConnectionManager
	manager = &ConnectionManager{
		clients:    make(map[*Client]bool),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		redis:      redisInstance,
		ctx:        context.Background(),
		tableSubscribers: &TableSubscribers{
			subscribers: make(map[string]map[*Client]bool),
		},
	}

	// Start ConnectionManager's main loop
	go manager.run()

	// Start Redis subscription loop
	go manager.subscribeToRedis()
}

// subscribeToRedis listens for database mutations from Redis and broadcasts them to clients
func (cm *ConnectionManager) subscribeToRedis() {
	// Check if ConnectionManager or Redis service is not properly initialized
	if cm == nil || cm.redis == nil {
		log.Println("Connection manager or Redis service not properly initialized")
		return
	}

	// Continuously subscribe to Redis channel for mutations
	for {
		pubsub := cm.redis.SubscribeToChannel("mutations")
		if pubsub == nil {
			log.Println("Failed to create Redis subscription, retrying in 5 seconds...")
			time.Sleep(5 * time.Second)
			continue
		}

		// Start receiving messages
		ch := pubsub.Channel()
		for msg := range ch {
			var broadcastMsg BroadcastMessage
			if err := json.Unmarshal([]byte(msg.Payload), &broadcastMsg); err != nil {
				log.Printf("Failed to unmarshal broadcast message: %v", err)
				continue
			}
			cm.broadcastToLocalClients(&broadcastMsg)
		}

		log.Println("Redis subscription channel closed, reconnecting...")
		pubsub.Close() // Close the subscription before reconnecting
		time.Sleep(time.Second)
	}
}

// Subscribe adds a client to a table's subscriber list
func (ts *TableSubscribers) Subscribe(table string, client *Client) {
	ts.mu.Lock()
	defer ts.mu.Unlock()
	
	if ts.subscribers[table] == nil {
		ts.subscribers[table] = make(map[*Client]bool)
	}
	ts.subscribers[table][client] = true
}

// Unsubscribe removes a client from a table's subscriber list
func (ts *TableSubscribers) Unsubscribe(table string, client *Client) {
	ts.mu.Lock()
	defer ts.mu.Unlock()
	
	if subs, exists := ts.subscribers[table]; exists {
		delete(subs, client)
		if len(subs) == 0 {
			delete(ts.subscribers, table)
		}
	}
}

// UnsubscribeAll removes a client from all tables
func (ts *TableSubscribers) UnsubscribeAll(client *Client) {
	ts.mu.Lock()
	defer ts.mu.Unlock()
	
	for table, subs := range ts.subscribers {
		delete(subs, client)
		if len(subs) == 0 {
			delete(ts.subscribers, table)
		}
	}
}

// GetSubscribers returns a copy of subscribers for a table
func (ts *TableSubscribers) GetSubscribers(table string) []*Client {
	ts.mu.RLock()
	defer ts.mu.RUnlock()
	
	if subs, exists := ts.subscribers[table]; exists {
		clients := make([]*Client, 0, len(subs))
		for client := range subs {
			clients = append(clients, client)
		}
		return clients
	}
	return nil
}

// broadcastToLocalClients sends mutation messages to all locally connected clients
// that are subscribed to the affected table
func (cm *ConnectionManager) broadcastToLocalClients(message *BroadcastMessage) {
	subscribers := cm.tableSubscribers.GetSubscribers(message.Table)
	if len(subscribers) == 0 {
		return
	}

	// Create the message once for all subscribers
	msg := WebSocketMessage{
		Type:   MessageTypeMutation,
		Tables: []string{message.Table},
		Data:   message.Data,
	}
	data, err := json.Marshal(msg)
	if err != nil {
		log.Printf("Failed to marshal message: %v", err)
		return
	}

	// Launch a single goroutine per table to handle all subscribers
	go func() {
		for _, client := range subscribers {
			client.mu.Lock()
			err := client.Conn.WriteMessage(websocket.TextMessage, data)
			client.mu.Unlock()

			if err != nil {
				cm.unregister <- client
			} else {
				cm.redis.IncrementMetric("messages_sent")
			}
		}
	}()
}

// Modified run function to handle subscriptions
func (cm *ConnectionManager) run() {
	for {
		select {
		case client := <-cm.register:
			cm.mu.Lock()
			cm.clients[client] = true
			cm.mu.Unlock()

			// Store session if enabled
			session := &redisService.SessionData{
				ClientID:    client.ID,
				Tables:      client.Tables,
				ConnectedAt: time.Now(),
			}
			if err := cm.redis.SaveSession(client.ID, session); err != nil {
				log.Printf("Failed to save session: %v", err)
			}

			// Subscribe to tables
			for table := range client.Tables {
				cm.tableSubscribers.Subscribe(table, client)
			}

			cm.redis.IncrementMetric("connections")
			client.sendStatus("Connected successfully")

		case client := <-cm.unregister:
			cm.mu.Lock()
			if _, ok := cm.clients[client]; ok {
				delete(cm.clients, client)
				cm.tableSubscribers.UnsubscribeAll(client)
				client.sendStatus("Disconnecting...")
				client.Conn.Close()

				for table := range client.Tables {
					cm.redis.RemoveSubscription(client.ID, table)
				}

				cm.redis.IncrementMetric("disconnections")
			}
			cm.mu.Unlock()
		}
	}
}

// BroadcastMutation publishes a table mutation to Redis for distribution to all servers
func BroadcastMutation(table string, body []byte) {
	// Prepare broadcast message
	msg := BroadcastMessage{
		Table: table,
		Data:  body,
	}

	// Publish message to Redis
	if err := manager.redis.PublishMessage("mutations", msg); err != nil {
		fmt.Printf("Redis publish error: %v\n", err)
		manager.redis.IncrementMetric("errors")
	}
}

// sendError sends an error message to the client
func (c *Client) sendError(err string) {
	// Prepare error message
	msg := WebSocketMessage{
		Type:  MessageTypeError,
		Error: err,
	}
	data, _ := json.Marshal(msg)
	c.Conn.WriteMessage(websocket.TextMessage, data)

	// Increment error metric
	manager.redis.IncrementMetric("errors")
}

// sendStatus sends a status message to the client
func (c *Client) sendStatus(status string) {
	// Prepare status message
	msg := WebSocketMessage{
		Type: MessageTypeStatus,
		Data: status,
	}
	data, _ := json.Marshal(msg)
	c.Conn.WriteMessage(websocket.TextMessage, data)
}

// getTableData retrieves data from a specific table and sends it to the client
func (c *Client) getTableData(table string) error {
	// Get database instance
	appState := app.GetInstance()
	db := appState.GetDB().GetDb()

	// Prepare query
	query := fmt.Sprintf("SELECT * FROM %s LIMIT 100", table)
	result, err := surrealdb.Query[any](db, query, nil)
	if err != nil {
		return err
	}

	// Prepare WebSocket message
	msg := WebSocketMessage{
		Type:   MessageTypeTableData,
		Tables: []string{table},
		Data:   result,
	}
	data, _ := json.Marshal(msg)
	return c.Conn.WriteMessage(websocket.TextMessage, data)
}

// generateClientID generates a unique identifier for a new client
func generateClientID() string {
	chars := "abcdefghijklmnopqrstuvwxyz0123456789"
	seed := rand.NewSource(time.Now().UnixNano())
	r := rand.New(seed)
	result := make([]byte, 32)
	for i := range result {
		result[i] = byte(r.Intn(len(chars)))
	}
	return string(result)
}

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true // Allow all origins for now, you may want to restrict this in production
	},
}

// HandleWebSocket handles WebSocket connections
func HandleWebSocket(c echo.Context) error {
	// Check if ConnectionManager is not properly initialized
	if manager == nil {
		return echo.NewHTTPError(http.StatusServiceUnavailable, "WebSocket service not available")
	}

	// Upgrade HTTP connection to WebSocket
	ws, err := upgrader.Upgrade(c.Response(), c.Request(), nil)
	if err != nil {
		return err
	}
	defer ws.Close()

	// Get user ID from cookie or header
	userId := getUserIdFromCookie(c)

	if userId == "" {
		userId = getUserIdFromHeader(c)

		if userId == "" {
			return echo.NewHTTPError(http.StatusUnauthorized, "Unauthorized")
		}
	}

	// Create new client
	client := &Client{
		ID:     generateClientID(),
		Conn:   ws,
		Tables: make(map[string]bool),
		userId: userId,
	}

	// Restore session if available
	if session, err := manager.redis.GetSession(client.ID); err == nil && session != nil {
		client.Tables = session.Tables
	}

	// Register new client
	manager.register <- client
	defer func() {
		manager.unregister <- client
	}()

	// Handle incoming messages
	for {
		messageType, message, err := ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("error: %v", err)
			}
			break
		}

		if messageType == websocket.TextMessage {
			// Check rate limit
			allowed, _ := manager.redis.CheckRateLimit(client.ID, MaxMessagesPerMinute, RateLimitWindow)
			if !allowed {
				client.sendError("Rate limit exceeded")
				continue
			}

			// Increment messages received metric
			manager.redis.IncrementMetric("messages_received")

			var msg WebSocketMessage
			if err := json.Unmarshal(message, &msg); err != nil {
				client.sendError("Invalid message format")
				continue
			}

			client.mu.Lock()
			switch msg.Type {
			case MessageTypeSubscribe:
				if len(msg.Tables) == 0 {
					client.sendError("No tables specified for subscription")
					client.mu.Unlock()
					continue
				}

				// TODO: Check if user is authorized to subscribe to these table

				for _, table := range msg.Tables {
					client.Tables[table] = true
					// Store subscription in Redis
					manager.redis.AddSubscription(client.ID, table)
					manager.redis.IncrementMetric("subscriptions")

					// Send initial table data
					if err := client.getTableData(table); err != nil {
						client.sendError(fmt.Sprintf("Failed to get data for table %s: %v", table, err))
					}
				}
				client.sendStatus(fmt.Sprintf("Subscribed to tables: %v", msg.Tables))

			case MessageTypeUnsubscribe:
				if len(msg.Tables) == 0 {
					client.sendError("No tables specified for unsubscription")
					client.mu.Unlock()
					continue
				}
				for _, table := range msg.Tables {
					delete(client.Tables, table)
					// Remove subscription from Redis
					manager.redis.RemoveSubscription(client.ID, table)
				}
				client.sendStatus(fmt.Sprintf("Unsubscribed from tables: %v", msg.Tables))

			case MessageTypeGetTableData:
				if len(msg.Tables) == 0 {
					client.sendError("No tables specified for data retrieval")
					client.mu.Unlock()
					continue
				}
				for _, table := range msg.Tables {
					if err := client.getTableData(table); err != nil {
						client.sendError(fmt.Sprintf("Failed to get data for table %s: %v", table, err))
					}
				}

			default:
				client.sendError("Unknown message type")
			}
			client.mu.Unlock()
		}
	}

	return nil
}

// getUserIdFromCookie retrieves the user ID from the cookie
func getUserIdFromCookie(c echo.Context) string {
	cookie, err := c.Cookie("user_id")
	if err != nil {
		return ""
	}
	return cookie.Value
}

// getUserIdFromHeader retrieves the user ID from the header
func getUserIdFromHeader(c echo.Context) string {
	return c.Request().Header.Get("user_id")
}

2. using go routine per client

// Package database implements WebSocket-based real-time database operations
package database

// MessageType represents different types of messages that can be exchanged between client and server
type MessageType string

// Constants defining various types of messages used in WebSocket communication
const (
	MessageTypeSubscribe    MessageType = "subscribe"    // Client requests to subscribe to table updates
	MessageTypeUnsubscribe  MessageType = "unsubscribe"  // Client requests to unsubscribe from table updates
	MessageTypeGetTableData MessageType = "get_table_data" // Client requests data from a specific table
	MessageTypeError        MessageType = "error"        // Server sends error messages to client
	MessageTypeStatus       MessageType = "status"       // Server sends status updates to client
	MessageTypeTableData    MessageType = "table_data"   // Server sends table data to client
	MessageTypeMutation     MessageType = "mutation"     // Server notifies clients about data changes
)

// Rate limiting constants to prevent abuse
const (
	MaxMessagesPerMinute = 100      // Maximum number of messages allowed per client per minute
	RateLimitWindow      = time.Minute // Time window for rate limiting
)

// WebSocketMessage defines the structure for all messages exchanged over WebSocket
type WebSocketMessage struct {
	Type   MessageType  `json:"type"`           // Type of the message
	Tables []string     `json:"tables,omitempty"` // List of tables (used for subscribe/unsubscribe)
	Data   interface{}  `json:"data,omitempty"`  // Payload data
	Error  string       `json:"error,omitempty"` // Error message if any
}

// Client represents a connected WebSocket client and its properties
type Client struct {
	ID     string          // Unique identifier for the client
	Conn   *websocket.Conn // WebSocket connection
	Tables map[string]bool // Map of tables this client is subscribed to
	mu     sync.RWMutex    // Mutex for thread-safe operations
	userId string          // User identifier for authentication
}

// ConnectionManager coordinates all WebSocket connections and message broadcasting
type ConnectionManager struct {
	clients    map[*Client]bool      // Map of all connected clients
	mu         sync.RWMutex          // Mutex for thread-safe operations
	register   chan *Client          // Channel for registering new clients
	unregister chan *Client          // Channel for unregistering clients
	redis      *redisService.RedisService // Redis service for pub/sub
	ctx        context.Context       // Context for cancellation
}

// BroadcastMessage represents a message to be broadcast to subscribed clients
type BroadcastMessage struct {
	Table string          `json:"table"` // Table that was modified
	Data  json.RawMessage `json:"data"`  // Modified data
}

var (
	manager *ConnectionManager
)

// init initializes the ConnectionManager and starts necessary background processes
func init() {
	// Initialize Redis service
	redisInstance, err := redisService.GetInstance()
	if err != nil {
		log.Printf("Failed to initialize Redis service: %v", err)
		return
	}

	// Initialize ConnectionManager
	manager = &ConnectionManager{
		clients:    make(map[*Client]bool),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		redis:      redisInstance,
		ctx:        context.Background(),
	}

	// Start ConnectionManager's main loop
	go manager.run()

	// Start Redis subscription loop
	go manager.subscribeToRedis()
}

// subscribeToRedis listens for database mutations from Redis and broadcasts them to clients
func (cm *ConnectionManager) subscribeToRedis() {
	// Check if ConnectionManager or Redis service is not properly initialized
	if cm == nil || cm.redis == nil {
		log.Println("Connection manager or Redis service not properly initialized")
		return
	}

	// Continuously subscribe to Redis channel for mutations
	for {
		pubsub := cm.redis.SubscribeToChannel("mutations")
		if pubsub == nil {
			log.Println("Failed to create Redis subscription, retrying in 5 seconds...")
			time.Sleep(5 * time.Second)
			continue
		}

		// Start receiving messages
		ch := pubsub.Channel()
		for msg := range ch {
			var broadcastMsg BroadcastMessage
			if err := json.Unmarshal([]byte(msg.Payload), &broadcastMsg); err != nil {
				log.Printf("Failed to unmarshal broadcast message: %v", err)
				continue
			}
			cm.broadcastToLocalClients(&broadcastMsg)
		}

		log.Println("Redis subscription channel closed, reconnecting...")
		pubsub.Close() // Close the subscription before reconnecting
		time.Sleep(time.Second)
	}
}

// broadcastToLocalClients sends mutation messages to all locally connected clients
// that are subscribed to the affected table
func (cm *ConnectionManager) broadcastToLocalClients(message *BroadcastMessage) {
	cm.mu.RLock()
	defer cm.mu.RUnlock()

	// Iterate over all connected clients
	for client := range cm.clients {
		client.mu.RLock()
		if _, ok := client.Tables[message.Table]; ok {
			go func(c *Client) {
				// Prepare WebSocket message
				msg := WebSocketMessage{
					Type:   MessageTypeMutation,
					Tables: []string{message.Table},
					Data:   message.Data,
				}
				data, _ := json.Marshal(msg)
				if err := c.Conn.WriteMessage(websocket.TextMessage, data); err != nil {
					cm.unregister <- c
				} else {
					// Increment messages sent metric
					cm.redis.IncrementMetric("messages_sent")
				}
			}(client)
		}
		client.mu.RUnlock()
	}
}

// run manages client connections and handles registration/unregistration
func (cm *ConnectionManager) run() {
	for {
		select {
		case client := <-cm.register:
			// Register new client
			cm.mu.Lock()
			cm.clients[client] = true
			cm.mu.Unlock()

			// Store session if enabled
			session := &redisService.SessionData{
				ClientID:    client.ID,
				Tables:      client.Tables,
				ConnectedAt: time.Now(),
			}
			if err := cm.redis.SaveSession(client.ID, session); err != nil {
				log.Printf("Failed to save session: %v", err)
			}

			// Increment connection metric
			cm.redis.IncrementMetric("connections")

			// Send initial status message to client
			client.sendStatus("Connected successfully")

		case client := <-cm.unregister:
			// Unregister client
			cm.mu.Lock()
			if _, ok := cm.clients[client]; ok {
				delete(cm.clients, client)
				client.sendStatus("Disconnecting...")
				client.Conn.Close()

				// Remove subscriptions from Redis
				for table := range client.Tables {
					cm.redis.RemoveSubscription(client.ID, table)
				}

				// Decrement connection metric
				cm.redis.IncrementMetric("disconnections")
			}
			cm.mu.Unlock()
		}
	}
}

// BroadcastMutation publishes a table mutation to Redis for distribution to all servers
func BroadcastMutation(table string, body []byte) {
	// Prepare broadcast message
	msg := BroadcastMessage{
		Table: table,
		Data:  body,
	}

	// Publish message to Redis
	if err := manager.redis.PublishMessage("mutations", msg); err != nil {
		fmt.Printf("Redis publish error: %v\n", err)
		manager.redis.IncrementMetric("errors")
	}
}

// sendError sends an error message to the client
func (c *Client) sendError(err string) {
	// Prepare error message
	msg := WebSocketMessage{
		Type:  MessageTypeError,
		Error: err,
	}
	data, _ := json.Marshal(msg)
	c.Conn.WriteMessage(websocket.TextMessage, data)

	// Increment error metric
	manager.redis.IncrementMetric("errors")
}

// sendStatus sends a status message to the client
func (c *Client) sendStatus(status string) {
	// Prepare status message
	msg := WebSocketMessage{
		Type: MessageTypeStatus,
		Data: status,
	}
	data, _ := json.Marshal(msg)
	c.Conn.WriteMessage(websocket.TextMessage, data)
}

// getTableData retrieves data from a specific table and sends it to the client
func (c *Client) getTableData(table string) error {
	// Get database instance
	appState := app.GetInstance()
	db := appState.GetDB().GetDb()

	// Prepare query
	query := fmt.Sprintf("SELECT * FROM %s LIMIT 100", table)
	result, err := surrealdb.Query[any](db, query, nil)
	if err != nil {
		return err
	}

	// Prepare WebSocket message
	msg := WebSocketMessage{
		Type:   MessageTypeTableData,
		Tables: []string{table},
		Data:   result,
	}
	data, _ := json.Marshal(msg)
	return c.Conn.WriteMessage(websocket.TextMessage, data)
}

// generateClientID generates a unique identifier for a new client
func generateClientID() string {
	chars := "abcdefghijklmnopqrstuvwxyz0123456789"
	seed := rand.NewSource(time.Now().UnixNano())
	r := rand.New(seed)
	result := make([]byte, 32)
	for i := range result {
		result[i] = byte(r.Intn(len(chars)))
	}
	return string(result)
}

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true // Allow all origins for now, you may want to restrict this in production
	},
}

// HandleWebSocket handles WebSocket connections
func HandleWebSocket(c echo.Context) error {
	// Check if ConnectionManager is not properly initialized
	if manager == nil {
		return echo.NewHTTPError(http.StatusServiceUnavailable, "WebSocket service not available")
	}

	// Upgrade HTTP connection to WebSocket
	ws, err := upgrader.Upgrade(c.Response(), c.Request(), nil)
	if err != nil {
		return err
	}
	defer ws.Close()

	// Get user ID from cookie or header
	userId := getUserIdFromCookie(c)

	if userId == "" {
		userId = getUserIdFromHeader(c)

		if userId == "" {
			return echo.NewHTTPError(http.StatusUnauthorized, "Unauthorized")
		}
	}

	// Create new client
	client := &Client{
		ID:     generateClientID(),
		Conn:   ws,
		Tables: make(map[string]bool),
		userId: userId,
	}

	// Restore session if available
	if session, err := manager.redis.GetSession(client.ID); err == nil && session != nil {
		client.Tables = session.Tables
	}

	// Register new client
	manager.register <- client
	defer func() {
		manager.unregister <- client
	}()

	// Handle incoming messages
	for {
		messageType, message, err := ws.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("error: %v", err)
			}
			break
		}

		if messageType == websocket.TextMessage {
			// Check rate limit
			allowed, _ := manager.redis.CheckRateLimit(client.ID, MaxMessagesPerMinute, RateLimitWindow)
			if !allowed {
				client.sendError("Rate limit exceeded")
				continue
			}

			// Increment messages received metric
			manager.redis.IncrementMetric("messages_received")

			var msg WebSocketMessage
			if err := json.Unmarshal(message, &msg); err != nil {
				client.sendError("Invalid message format")
				continue
			}

			client.mu.Lock()
			switch msg.Type {
			case MessageTypeSubscribe:
				if len(msg.Tables) == 0 {
					client.sendError("No tables specified for subscription")
					client.mu.Unlock()
					continue
				}

				// TODO: Check if user is authorized to subscribe to these table

				for _, table := range msg.Tables {
					client.Tables[table] = true
					// Store subscription in Redis
					manager.redis.AddSubscription(client.ID, table)
					manager.redis.IncrementMetric("subscriptions")

					// Send initial table data
					if err := client.getTableData(table); err != nil {
						client.sendError(fmt.Sprintf("Failed to get data for table %s: %v", table, err))
					}
				}
				client.sendStatus(fmt.Sprintf("Subscribed to tables: %v", msg.Tables))

			case MessageTypeUnsubscribe:
				if len(msg.Tables) == 0 {
					client.sendError("No tables specified for unsubscription")
					client.mu.Unlock()
					continue
				}
				for _, table := range msg.Tables {
					delete(client.Tables, table)
					// Remove subscription from Redis
					manager.redis.RemoveSubscription(client.ID, table)
				}
				client.sendStatus(fmt.Sprintf("Unsubscribed from tables: %v", msg.Tables))

			case MessageTypeGetTableData:
				if len(msg.Tables) == 0 {
					client.sendError("No tables specified for data retrieval")
					client.mu.Unlock()
					continue
				}
				for _, table := range msg.Tables {
					if err := client.getTableData(table); err != nil {
						client.sendError(fmt.Sprintf("Failed to get data for table %s: %v", table, err))
					}
				}

			default:
				client.sendError("Unknown message type")
			}
			client.mu.Unlock()
		}
	}

	return nil
}

// getUserIdFromCookie retrieves the user ID from the cookie
func getUserIdFromCookie(c echo.Context) string {
	cookie, err := c.Cookie("user_id")
	if err != nil {
		return ""
	}
	return cookie.Value
}

// getUserIdFromHeader retrieves the user ID from the header
func getUserIdFromHeader(c echo.Context) string {
	return c.Request().Header.Get("user_id")
}

Also am using echo as my web server

[FZambia]

Hello @SOG-web

I am author of https://github.com/centrifugal/centrifuge and https://github.com/centrifugal/centrifugo and the thing you are doing is very similar to what those projects provide and to what I've spent enough of my time for. Basically: multiplexing of subscriptions over client connection, scalability over Redis and even integration with PostgreSQL outbox (which looks very similar to your mutations stream goal).

I am not sure whether you are doing this for educational purposes or not. If yes – maybe take a look at source code of projects I pointed and probably you will find some useful ideas. If you are developing for non-educational – then you can just use Centrifuge library (it uses a recent fork of Gorilla WebSocket btw) or Centrifugo (which is built on top of Centrifuge library and provides more features at the cost of being self-contained standalone server with its own rules and mechanics).

Now closer to your question. I'd say that broadcasting messages each in a separate goroutine is not a good approach. I'd suggest having 2 goroutines per connection. One reads messages from the connection, and another writes message to the connection by reading from a per-connection queue (a-la mailbox). This allows to achieve reasonably fast broadcasts (where you just put the message into individual connection queue) without making a separate goroutine for each message, which looks rather unpredictable.