Documentation
¶
Overview ¶
Package websocket provides WebSocket connection management.
Package websocket provides WebSocket connection management.
Index ¶
- Variables
- func AcceptHandler(mgr *ConnectionManager, cfg config.WebSocketConfig, logger *slog.Logger) http.HandlerFunc
- func CheckSDKVersion(r *http.Request, minVersion int) error
- type ClientResponse
- type Connection
- func (c *Connection) Close(ctx context.Context) error
- func (c *Connection) CloseNow() error
- func (c *Connection) ConnID() string
- func (c *Connection) CreatedAt() time.Time
- func (c *Connection) Done() <-chan struct{}
- func (c *Connection) NotifyClose() error
- func (c *Connection) Send(data []byte) error
- func (c *Connection) SetMetrics(m *telemetry.Metrics)
- func (c *Connection) SetRateLimiter(msgPerSec float64, burst int)
- func (c *Connection) Start(ctx context.Context, ...)
- func (c *Connection) SubscribeChannel(channel string) error
- func (c *Connection) SubscribedChannels() map[string]struct{}
- func (c *Connection) UnsubscribeChannel(channel string)
- func (c *Connection) UserID() string
- type ConnectionManager
- func (m *ConnectionManager) BroadcastToUser(_ context.Context, userID string, data []byte)
- func (m *ConnectionManager) Close(drainTimeout time.Duration) error
- func (m *ConnectionManager) Get(connID string) (*Connection, bool)
- func (m *ConnectionManager) GetByUser(userID string) []*Connection
- func (m *ConnectionManager) GetMaxSeq(ctx context.Context, conn *Connection, requestID string) []byte
- func (m *ConnectionManager) HandleClientResponse(resp *ClientResponse) bool
- func (m *ConnectionManager) NewConnID() string
- func (m *ConnectionManager) Publish(ctx context.Context, channel string, data []byte) error
- func (m *ConnectionManager) Register(_ context.Context, conn *Connection) error
- func (m *ConnectionManager) ReleaseConn()
- func (m *ConnectionManager) RouteToConn(_ context.Context, connID string, data []byte) error
- func (m *ConnectionManager) SendRequestToClient(ctx context.Context, connID string, method string, payload []byte, ...) (*ClientResponse, error)
- func (m *ConnectionManager) SetCloseHandler(h func(ctx context.Context, connID string))
- func (m *ConnectionManager) SetMessageHandler(h func(ctx context.Context, connID string, data []byte))
- func (m *ConnectionManager) Start(ctx context.Context)
- func (m *ConnectionManager) Subscribe(ctx context.Context, connID, channel string) error
- func (m *ConnectionManager) TotalConns() int64
- func (m *ConnectionManager) TryReserveConn() (int64, bool)
- func (m *ConnectionManager) Unregister(connID string)
- func (m *ConnectionManager) Unsubscribe(connID, channel string)
- type Option
- func WithAPIKey(key string) Option
- func WithCache(c cache.Cache) Option
- func WithCloseHandler(h func(ctx context.Context, connID string)) Option
- func WithConnRenewInterval(d time.Duration) Option
- func WithGracePeriod(d time.Duration) Option
- func WithLogger(l *slog.Logger) Option
- func WithMaxTotalConns(maxConns int64) Option
- func WithMessageHandler(h func(ctx context.Context, connID string, data []byte)) Option
- func WithMetrics(m *telemetry.Metrics) Option
Constants ¶
This section is empty.
Variables ¶
var ( // ErrClosed is returned when Send is called on a closed connection. ErrClosed = errors.New("websocket: connection closed") // ErrSendFull is returned when the send buffer is full. ErrSendFull = errors.New("websocket: send buffer full") // ErrInvalidConnID is returned when NewConnection receives an empty connID. ErrInvalidConnID = errors.New("websocket: connID must not be empty") // ErrInvalidUserID is returned when NewConnection receives an empty or too-long userID. ErrInvalidUserID = errors.New("websocket: userID must not be empty or exceed 256 characters") // ErrEmptyChannel is returned when SubscribeChannel receives an empty channel name. ErrEmptyChannel = errors.New("websocket: channel name must not be empty") )
var ErrAlreadyRegistered = errors.New("websocket: connection already registered")
ErrAlreadyRegistered is returned when Register is called with a duplicate connID.
var ErrMaxTotalConns = errors.New("websocket: maximum total connections reached")
ErrMaxTotalConns is returned when the global connection limit is reached.
var ErrNotFound = errors.New("websocket: connection not found")
ErrNotFound is returned when a connection lookup fails.
Functions ¶
func AcceptHandler ¶
func AcceptHandler(mgr *ConnectionManager, cfg config.WebSocketConfig, logger *slog.Logger) http.HandlerFunc
AcceptHandler returns an http.HandlerFunc that upgrades HTTP to WebSocket. Flow: authenticate (if API key configured) -> read X-Sequify-User-ID header -> check SDK version -> check connection limit -> accept -> register -> start.
Security: if mgr.apiKey is non-empty, every client must present the key via "Authorization: Bearer <key>" or "X-API-Key: <key>" header. The comparison uses crypto/subtle.ConstantTimeCompare to prevent timing side-channels. When apiKey is empty (development mode), authentication is disabled.
func CheckSDKVersion ¶
CheckSDKVersion validates the X-Sequify-SDK-Version header against the minimum version. Returns nil if compatible, or an errcode error (4006) if incompatible. Missing header is treated as version 0. Supports both integer ("1") and semver ("1.1.0") formats — extracts major version.
Types ¶
type ClientResponse ¶
type ClientResponse struct {
ID string `json:"id"`
Code int `json:"code"`
Message string `json:"message"`
Payload json.RawMessage `json:"payload,omitempty"`
}
ClientResponse represents a response from a client to a server-initiated request.
type Connection ¶
type Connection struct {
// contains filtered or unexported fields
}
Connection wraps a coder/websocket.Conn with goroutine-safe send channel and idempotent close lifecycle.
func NewConnection ¶
func NewConnection(connID, userID string, conn *websocket.Conn, logger *slog.Logger) (*Connection, error)
NewConnection creates a new Connection wrapping the given websocket.Conn. The connID and userID are set at creation and never mutated. The connection context is initialized to context.Background() and is immutable. Callers should pass a separate ctx to Start() for goroutine lifecycle.
Returns an error if connID is empty or userID is empty/too long.
func (*Connection) Close ¶
func (c *Connection) Close(ctx context.Context) error
Close idempotently closes the connection. Sequence: close(writeCh) -> conn.Close(StatusNormalClosure) -> conn.CloseNow() -> close(closeCh). Returns nil (close errors are best-effort).
func (*Connection) CloseNow ¶
func (c *Connection) CloseNow() error
CloseNow immediately closes the connection without graceful handshake. Used for force-close during drain timeout. Idempotent.
func (*Connection) CreatedAt ¶
func (c *Connection) CreatedAt() time.Time
CreatedAt returns the time the connection was created.
func (*Connection) Done ¶
func (c *Connection) Done() <-chan struct{}
Done returns a channel that is closed when the connection is fully closed.
func (*Connection) NotifyClose ¶
func (c *Connection) NotifyClose() error
NotifyClose sends a close frame to the client without closing the underlying connection. Used during graceful shutdown to notify clients before draining. Safe to call on already-closed connections (returns nil).
func (*Connection) Send ¶
func (c *Connection) Send(data []byte) error
Send writes data to the connection's send channel. Returns ErrClosed if the connection is closed. Returns ErrSendFull if the send buffer is full.
func (*Connection) SetMetrics ¶
func (c *Connection) SetMetrics(m *telemetry.Metrics)
SetMetrics attaches Prometheus metrics for recording message counts. Call before Start().
func (*Connection) SetRateLimiter ¶
func (c *Connection) SetRateLimiter(msgPerSec float64, burst int)
SetRateLimiter configures a per-connection message rate limiter. When set, readPump will close the connection with StatusPolicyViolation if the client exceeds the configured rate. Call before Start(); safe to skip (nil = disabled).
func (*Connection) Start ¶
func (c *Connection) Start(ctx context.Context, handler func(ctx context.Context, connID string, data []byte), pingInterval, pongTimeout time.Duration)
Start launches the writePump and heartbeat goroutines, then blocks in readPump. The handler is called for each text message received. This method blocks until the connection is closed. CRITICAL: ctx should be a fresh context, NOT r.Context() from Accept. The ctx parameter is for goroutine lifecycle only — it is not stored in the Connection struct.
func (*Connection) SubscribeChannel ¶
func (c *Connection) SubscribeChannel(channel string) error
SubscribeChannel adds a channel to this connection's subscription set. Returns ErrEmptyChannel if the channel name is empty.
func (*Connection) SubscribedChannels ¶
func (c *Connection) SubscribedChannels() map[string]struct{}
SubscribedChannels returns a snapshot copy of this connection's subscribed channels.
func (*Connection) UnsubscribeChannel ¶
func (c *Connection) UnsubscribeChannel(channel string)
UnsubscribeChannel removes a channel from this connection's subscription set.
type ConnectionManager ¶
type ConnectionManager struct {
// contains filtered or unexported fields
}
ConnectionManager maintains a concurrent-safe registry of all active connections. It provides O(1) lookup by connID and userID.
func NewConnectionManager ¶
func NewConnectionManager(opts ...Option) *ConnectionManager
NewConnectionManager creates a new ConnectionManager with the given options.
func (*ConnectionManager) BroadcastToUser ¶
func (m *ConnectionManager) BroadcastToUser(_ context.Context, userID string, data []byte)
BroadcastToUser sends data to all connections of a user. Logs on individual Send failures but continues to other connections.
func (*ConnectionManager) Close ¶
func (m *ConnectionManager) Close(drainTimeout time.Duration) error
Close performs a graceful shutdown with configurable drain timeout. Flow: stop renewal timer -> cancel server context -> stop grace timers -> cancel Redis subs -> wait callbacks -> drain connections -> force close.
func (*ConnectionManager) Get ¶
func (m *ConnectionManager) Get(connID string) (*Connection, bool)
Get returns the connection for the given connID. Returns (conn, true) if found, (nil, false) otherwise.
func (*ConnectionManager) GetByUser ¶
func (m *ConnectionManager) GetByUser(userID string) []*Connection
GetByUser returns all connections for a user. Returns an empty slice (not nil) if no connections.
func (*ConnectionManager) GetMaxSeq ¶
func (m *ConnectionManager) GetMaxSeq(ctx context.Context, conn *Connection, requestID string) []byte
GetMaxSeq returns the current max sequence numbers for the connection's channels. Returns serialized Response bytes ready to send via conn.Send(). On cache error, returns an error response (does NOT return error to caller).
Only the user channel is returned. All updates (including conversation replies and tool calls) are stored at user scope after the write-fanout design change: the server publishes to each conversation member's user channel, so clients only need to track one seq stream.
func (*ConnectionManager) HandleClientResponse ¶
func (m *ConnectionManager) HandleClientResponse(resp *ClientResponse) bool
HandleClientResponse matches an incoming client response to a pending request and delivers it via the pending channel. Returns true if a matching request was found. Should be called from the message router when a Response type message arrives.
func (*ConnectionManager) NewConnID ¶
func (m *ConnectionManager) NewConnID() string
NewConnID generates a new connection ID using idgen.New().
func (*ConnectionManager) Publish ¶
Publish sends data to the specified channel via the cache's Pub/Sub. Returns an error if the cache is not configured or publish fails.
func (*ConnectionManager) Register ¶
func (m *ConnectionManager) Register(_ context.Context, conn *Connection) error
Register adds a connection to the manager. Returns error if the connection is already registered (duplicate connID).
func (*ConnectionManager) ReleaseConn ¶
func (m *ConnectionManager) ReleaseConn()
ReleaseConn decrements the global connection count.
func (*ConnectionManager) RouteToConn ¶
RouteToConn sends data to a specific connection. Returns ErrNotFound if the connection does not exist. Returns ErrClosed if the target connection is closed.
func (*ConnectionManager) SendRequestToClient ¶
func (m *ConnectionManager) SendRequestToClient( ctx context.Context, connID string, method string, payload []byte, timeout time.Duration, ) (*ClientResponse, error)
SendRequestToClient sends a request to a specific connection and blocks until the client responds or the timeout expires. The request ID is generated using idgen.New() for global uniqueness. Returns the client's response or an error (timeout, send failure, context cancelled).
func (*ConnectionManager) SetCloseHandler ¶
func (m *ConnectionManager) SetCloseHandler(h func(ctx context.Context, connID string))
SetCloseHandler sets the callback invoked when a connection closes. This enables late-binding of the session cleanup handler after ConnectionManager creation.
func (*ConnectionManager) SetMessageHandler ¶
func (m *ConnectionManager) SetMessageHandler(h func(ctx context.Context, connID string, data []byte))
SetMessageHandler sets the message handler for incoming WebSocket messages. This enables late-binding of the session handler after ConnectionManager creation.
func (*ConnectionManager) Start ¶
func (m *ConnectionManager) Start(ctx context.Context)
Start begins background goroutines (renewal loop). Must be called after NewConnectionManager and before accepting connections. The context is used for server-level operations like Redis subscriptions.
func (*ConnectionManager) Subscribe ¶
func (m *ConnectionManager) Subscribe(ctx context.Context, connID, channel string) error
Subscribe adds a channel subscription for the given connection. Writes to both the manager-side index and the connection-side set. Returns ErrNotFound if the connection does not exist. Idempotent: subscribing to an already-subscribed channel is a no-op. If this is the first local subscriber to the channel and a cache is configured, a Redis Pub/Sub subscription is started.
func (*ConnectionManager) TotalConns ¶
func (m *ConnectionManager) TotalConns() int64
TotalConns returns the current total number of active connections.
func (*ConnectionManager) TryReserveConn ¶
func (m *ConnectionManager) TryReserveConn() (int64, bool)
TryReserveConn atomically increments the global connection count if under the limit. Returns (newCount, true) if a slot was reserved, (0, false) if the limit was reached.
func (*ConnectionManager) Unregister ¶
func (m *ConnectionManager) Unregister(connID string)
Unregister removes a connection from the manager. Safe to call with non-existent connID (no-op).
func (*ConnectionManager) Unsubscribe ¶
func (m *ConnectionManager) Unsubscribe(connID, channel string)
Unsubscribe removes a channel subscription for the given connection. Safe to call with non-existent connID or channel (no-op). If this was the last subscriber, starts the grace period.
type Option ¶
type Option func(*options)
Option configures ConnectionManager.
func WithAPIKey ¶
WithAPIKey sets the API key required for WebSocket connections. If empty (default), authentication is disabled (development mode). If non-empty, clients must send "Authorization: Bearer <key>" or "X-API-Key: <key>" headers.
func WithCloseHandler ¶
WithCloseHandler sets a callback invoked when a connection closes. Used by the session layer to clean up per-connection state (e.g., pending fetches).
func WithConnRenewInterval ¶
WithConnRenewInterval sets the interval for batch TTL renewal of connection counts.
func WithGracePeriod ¶
WithGracePeriod sets the grace period before unsubscribing from a channel after the last local subscriber leaves. This allows time for new subscribers to join without losing the Redis subscription.
func WithLogger ¶
WithLogger sets the logger on ConnectionManager.
func WithMaxTotalConns ¶
WithMaxTotalConns sets the global connection limit. When the limit is reached, new connections are rejected with 503. Default: 10000. Set to 0 to disable (not recommended for production).
func WithMessageHandler ¶
WithMessageHandler sets a custom message handler for incoming WebSocket messages. If not set, messages are logged and dropped.
func WithMetrics ¶
WithMetrics sets the Prometheus metrics for recording connection and message counts. If nil, metrics recording is disabled.
Source Files