job-search/internal/websocket/hub.go

package websocket

import (
	"context"
	"encoding/json"
	"log/slog"
	"sync"

	ws "github.com/coder/websocket"

	"git.avsdev.uk/Go/job-search/internal/config"
)

// Hub tracks connected clients, WebSocket (ServeHTTP) and plain TCP
// (ServeTCP) alike, and sends pushed messages to all of them.
type Hub struct {
	conf config.JSCServerWebSocket
	log  *slog.Logger

	m       sync.RWMutex
	clients map[*client]struct{}
	closed  bool
	wg      sync.WaitGroup
}

// defaultMaxMessageSize is the largest message read from a client when
// max_message_size is not set.
const defaultMaxMessageSize = 32 << 10

// NewHub creates a hub.
func NewHub(conf config.JSCServerWebSocket, log *slog.Logger) *Hub {
	return &Hub{
		conf:    conf,
		log:     log.With("component", "ws"),
		clients: make(map[*client]struct{}),
	}
}

// Push sends a message of type msgType to every client. The payload is
// marshalled once and the same bytes are queued for every client; a client
// whose queue is full is closed as a slow consumer. A payload that can't be
// marshalled is logged and nothing is sent.
func (h *Hub) Push(msgType string, payload any) {
	var rawPayload json.RawMessage
	if payload != nil {
		var err error
		if rawPayload, err = json.Marshal(payload); err != nil {
			h.log.Error("marshal push payload", "type", msgType, "error", err)
			return
		}
	}

	msg, err := json.Marshal(Envelope{
		Type:    msgType,
		Payload: rawPayload,
	})
	if err != nil {
		h.log.Error("marshal push", "type", msgType, "error", err)
		return
	}

	var slow []*client
	h.m.RLock()
	for client := range h.clients {
		if !client.trySend(msg) {
			slow = append(slow, client)
		}
	}
	h.m.RUnlock()

	for _, client := range slow {
		h.log.Warn("dropping slow client", "client", client.id, "ip", client.ip)
		go client.close(ws.StatusPolicyViolation, "slow consumer")
	}
}

// Count returns the number of connected clients.
func (h *Hub) Count() int {
	h.m.RLock()
	defer h.m.RUnlock()
	return len(h.clients)
}

// Shutdown closes every connection with 1001 (going away) and waits for the
// handlers to return. If ctx expires first, the remaining connections are
// closed at once and ctx's error is returned. New connections are refused
// afterwards.
func (h *Hub) Shutdown(ctx context.Context) error {
	h.m.Lock()
	h.closed = true
	clientShutdowns := make([]*client, 0, len(h.clients))
	for client := range h.clients {
		clientShutdowns = append(clientShutdowns, client)
	}
	h.m.Unlock()

	if len(clientShutdowns) == 0 {
		// Nothing to close. Clients that just left are only finishing their
		// unregister, so this wait is short and ctx doesn't apply.
		h.wg.Wait()
		return nil
	}

	for _, client := range clientShutdowns {
		go client.close(ws.StatusGoingAway, "server shutting down")
	}

	done := make(chan struct{})
	go func() {
		h.wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		return nil
	case <-ctx.Done():
		// Both may be ready; a finished shutdown is not a timeout.
		select {
		case <-done:
			return nil
		default:
		}
		for _, client := range clientShutdowns {
			client.conn.CloseNow() //nolint:errcheck,gosec // out of time; dropping it
		}
		return ctx.Err()
	}
}

// maxMessageSize is the largest message read from a client: max_message_size,
// or defaultMaxMessageSize if that is 0 or less.
func (h *Hub) maxMessageSize() int {
	if h.conf.MaxMessageSize <= 0 {
		return defaultMaxMessageSize
	}
	return int(h.conf.MaxMessageSize)
}

// register adds a client to the hub. Hub clients receive all push messages and
// have their lifecycles tracked by the hub.
func (h *Hub) register(c *client) bool {
	h.m.Lock()
	defer h.m.Unlock()
	if h.closed {
		return false
	}
	h.clients[c] = struct{}{}
	h.wg.Add(1)
	return true
}

// unregister removes a client from the hub.
func (h *Hub) unregister(c *client) {
	h.m.Lock()
	delete(h.clients, c)
	h.m.Unlock()
	h.wg.Done()
}