socket

package
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 3, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func NotificationsStream

func NotificationsStream(userID string) presence.StreamKey

NotificationsStream returns the stream key for a user's notification inbox.

Types

type ChannelJoinPayload

type ChannelJoinPayload struct {
	Target      string `json:"target"`
	Type        int    `json:"type"` // 1=ROOM, 2=DM, 3=GROUP
	Persistence *bool  `json:"persistence,omitempty"`
	Hidden      *bool  `json:"hidden,omitempty"`
}

type ChannelLeavePayload

type ChannelLeavePayload struct {
	ChannelID string `json:"channel_id"`
}

type ChannelMessageRemovePayload

type ChannelMessageRemovePayload struct {
	ChannelID string `json:"channel_id"`
	MessageID string `json:"message_id"`
}

type ChannelMessageSendPayload

type ChannelMessageSendPayload struct {
	ChannelID string          `json:"channel_id"`
	Content   json.RawMessage `json:"content"`
}

type ChannelMessageUpdatePayload

type ChannelMessageUpdatePayload struct {
	ChannelID string          `json:"channel_id"`
	MessageID string          `json:"message_id"`
	Content   json.RawMessage `json:"content"`
}

type ConnectionRegistry

type ConnectionRegistry struct {
	GracePeriod time.Duration // Reconnection grace period
	// contains filtered or unexported fields
}

ConnectionRegistry maintains active and recovering user WebSocket sessions.

func NewConnectionRegistry

func NewConnectionRegistry() *ConnectionRegistry

NewConnectionRegistry creates a new ConnectionRegistry.

func (*ConnectionRegistry) ActiveSessionCount

func (cr *ConnectionRegistry) ActiveSessionCount() int

ActiveSessionCount returns sessions currently tracked (including grace-period recovery).

func (*ConnectionRegistry) Add

func (cr *ConnectionRegistry) Add(s *Session)

Add registers a new session, cancelling any active grace timers for reconnection.

func (*ConnectionRegistry) AllSessions

func (cr *ConnectionRegistry) AllSessions() []*Session

AllSessions returns a snapshot of all active sessions.

func (*ConnectionRegistry) GetBySession

func (cr *ConnectionRegistry) GetBySession(sessionID string) (*Session, bool)

GetBySession retrieves a session by ID.

func (*ConnectionRegistry) GetUserSessionIDs

func (cr *ConnectionRegistry) GetUserSessionIDs(userID string) []string

GetUserSessionIDs retrieves active session IDs for a user.

func (*ConnectionRegistry) GetUserSessions

func (cr *ConnectionRegistry) GetUserSessions(userID string) []*Session

GetUserSessions returns active sessions for a user.

func (*ConnectionRegistry) SendToSession

func (cr *ConnectionRegistry) SendToSession(sessionID string, payload []byte)

SendToSession sends a message to a specific active session. When the send buffer is full the session is closed (backpressure).

func (*ConnectionRegistry) SendToUser

func (cr *ConnectionRegistry) SendToUser(userID string, payload []byte)

SendToUser sends a message to all active sessions of a user. When a session send buffer is full that session is closed (backpressure).

func (*ConnectionRegistry) StartGracePeriod

func (cr *ConnectionRegistry) StartGracePeriod(sessionID string, cleanupFn func())

StartGracePeriod kicks off the connection recovery window.

type Envelope

type Envelope struct {
	Cid                   string                        `json:"cid,omitempty"`
	MatchCreate           *MatchCreatePayload           `json:"match_create,omitempty"`
	MatchJoin             *MatchJoinPayload             `json:"match_join,omitempty"`
	MatchLeave            *MatchLeavePayload            `json:"match_leave,omitempty"`
	MatchDataSend         *MatchDataSendPayload         `json:"match_data_send,omitempty"`
	MatchmakerAdd         *MatchmakerAddPayload         `json:"matchmaker_add,omitempty"`
	MatchmakerRemove      *MatchmakerRemovePayload      `json:"matchmaker_remove,omitempty"`
	PartyMatchmakerAdd    *PartyMatchmakerAddPayload    `json:"party_matchmaker_add,omitempty"`
	PartyMatchmakerRemove *PartyMatchmakerRemovePayload `json:"party_matchmaker_remove,omitempty"`
	PartyCreate           *PartyCreatePayload           `json:"party_create,omitempty"`
	PartyJoin             *PartyJoinPayload             `json:"party_join,omitempty"`
	PartyLeave            *PartyLeavePayload            `json:"party_leave,omitempty"`
	PartyPromote          *PartyPromotePayload          `json:"party_promote,omitempty"`
	PartyAccept           *PartyAcceptPayload           `json:"party_accept,omitempty"`
	PartyRemove           *PartyRemovePayload           `json:"party_remove,omitempty"`
	PartyClose            *PartyClosePayload            `json:"party_close,omitempty"`
	PartyUpdate           *PartyUpdatePayload           `json:"party_update,omitempty"`
	PartyJoinRequestList  *PartyJoinRequestListPayload  `json:"party_join_request_list,omitempty"`
	PartyDataSend         *PartyDataSendPayload         `json:"party_data_send,omitempty"`
	StatusFollow          *StatusFollowPayload          `json:"status_follow,omitempty"`
	StatusUnfollow        *StatusUnfollowPayload        `json:"status_unfollow,omitempty"`
	StatusUpdate          *StatusUpdatePayload          `json:"status_update,omitempty"`
	Ping                  *PingPayload                  `json:"ping,omitempty"`
	ChannelJoin           *ChannelJoinPayload           `json:"channel_join,omitempty"`
	ChannelLeave          *ChannelLeavePayload          `json:"channel_leave,omitempty"`
	ChannelMessageSend    *ChannelMessageSendPayload    `json:"channel_message_send,omitempty"`
	ChannelMessageUpdate  *ChannelMessageUpdatePayload  `json:"channel_message_update,omitempty"`
	ChannelMessageRemove  *ChannelMessageRemovePayload  `json:"channel_message_remove,omitempty"`
	Rpc                   *RpcPayload                   `json:"rpc,omitempty"`
}

Envelope defines client-to-server and server-to-client WS structures.

type GatewayHandler

type GatewayHandler struct {
	Router           *match.Router
	Matchmaker       *matchmaker.Matchmaker
	PartyRegistry    *party.Registry
	HookRegistry     *runtime.HookRegistry
	PresenceTracker  *presence.PresenceTracker
	StreamTracker    *presence.StreamTracker
	StatusRegistry   *presence.StatusRegistry
	MessageRouter    *presence.LocalMessageRouter
	MaxStatusBytes   int
	MaxSubscriptions int
	// contains filtered or unexported fields
}

GatewayHandler upgrades connections and starts connection lifecycle loops.

func NewGatewayHandler

func NewGatewayHandler(
	logger *zap.Logger,
	tm *auth.TokenManager,
	reg *ConnectionRegistry,
	onConnect func(s *Session),
	onDisconnect func(sessionID, userID, username string),
	router *match.Router,
) *GatewayHandler

NewGatewayHandler creates a new GatewayHandler.

func (*GatewayHandler) BroadcastChannelMessage

func (gh *GatewayHandler) BroadcastChannelMessage(channelID string, msg *chat.ChannelMessage)

BroadcastChannelMessage implements chat.MessageRouter for local WS fan-out.

func (*GatewayHandler) BroadcastGroupChannelMessage

func (gh *GatewayHandler) BroadcastGroupChannelMessage(ctx context.Context, groupID string, code int16, content, senderID, username string)

BroadcastGroupChannelMessage implements social.ChannelBroadcaster.

func (*GatewayHandler) CleanupRelayedMatch

func (gh *GatewayHandler) CleanupRelayedMatch(matchID string)

CleanupRelayedMatch deletes the relayed match state.

func (*GatewayHandler) DeliverClusterChat

func (gh *GatewayHandler) DeliverClusterChat(channelID string, payload []byte)

DeliverClusterChat delivers a cross-node chat payload to local channel subscribers.

func (*GatewayHandler) DeliverClusterParty

func (gh *GatewayHandler) DeliverClusterParty(partyID string, payload []byte)

DeliverClusterParty delivers a remote party envelope to local party members.

func (*GatewayHandler) DeliverClusterPresence

func (gh *GatewayHandler) DeliverClusterPresence(kind string, payload []byte)

DeliverClusterPresence applies a remote status join/leave/update to local followers.

func (*GatewayHandler) DisconnectSession

func (gh *GatewayHandler) DisconnectSession(sessionID string) error

DisconnectSession force-closes a session (runtime SessionDisconnect).

func (*GatewayHandler) EmitSessionEvent

func (gh *GatewayHandler) EmitSessionEvent(name, userID, sessionID, username string)

EmitSessionEvent dispatches session_start / session_end via configured dispatcher.

func (*GatewayHandler) RouteMessage

func (gh *GatewayHandler) RouteMessage(s *Session, payload []byte)

func (*GatewayHandler) SendNotifications

func (gh *GatewayHandler) SendNotifications(userID string, notifs []*notification.Notification)

SendNotifications implements notification.Deliverer for a single user.

func (*GatewayHandler) SendNotificationsToAll

func (gh *GatewayHandler) SendNotificationsToAll(notifs []*notification.Notification)

SendNotificationsToAll implements notification.Deliverer for connected clients.

func (*GatewayHandler) SetClusterMesh

func (gh *GatewayHandler) SetClusterMesh(mesh *cluster.Mesh)

SetClusterMesh attaches the Redis social mesh for chat/presence/party fan-out.

func (*GatewayHandler) SetDBPool

func (gh *GatewayHandler) SetDBPool(pool *pgxpool.Pool)

SetDBPool configures an optional Postgres pool for chat persistence.

func (*GatewayHandler) SetEventDispatcher

func (gh *GatewayHandler) SetEventDispatcher(fn func(name, userID, sessionID, username string))

SetEventDispatcher configures session_start/session_end event emission.

func (*GatewayHandler) SetHookRegistry

func (gh *GatewayHandler) SetHookRegistry(hr *runtime.HookRegistry)

SetHookRegistry configures runtime hooks for realtime interceptors.

func (*GatewayHandler) SetMatchmaker

func (gh *GatewayHandler) SetMatchmaker(mm *matchmaker.Matchmaker)

SetMatchmaker configures the Matchmaker instance for ticket routing.

func (*GatewayHandler) SetMessageRouter

func (gh *GatewayHandler) SetMessageRouter(r *presence.LocalMessageRouter)

SetMessageRouter configures the local message router for status delivery.

func (*GatewayHandler) SetNodeID

func (gh *GatewayHandler) SetNodeID(nodeID string)

SetNodeID configures the cluster node label used for match metadata and Pub/Sub.

func (*GatewayHandler) SetPartyRegistry

func (gh *GatewayHandler) SetPartyRegistry(reg *party.Registry)

SetPartyRegistry configures the party registry for party matchmaker add/remove.

func (*GatewayHandler) SetPresenceLimits

func (gh *GatewayHandler) SetPresenceLimits(maxStatusBytes, maxSubscriptions int)

SetPresenceLimits configures status string and follow caps.

func (*GatewayHandler) SetPresenceTracker

func (gh *GatewayHandler) SetPresenceTracker(pt *presence.PresenceTracker)

SetPresenceTracker configures the online index for friends∩online (legacy field name).

func (*GatewayHandler) SetRPCInvoker

func (gh *GatewayHandler) SetRPCInvoker(fn func(ctx context.Context, userID, username, id, payload string) (string, int, error))

SetRPCInvoker configures WebSocket custom RPC dispatch.

func (*GatewayHandler) SetRedisClient

func (gh *GatewayHandler) SetRedisClient(rdb *redis.Client)

SetRedisClient configures the Redis client instance.

func (*GatewayHandler) SetRtHookExecutor

func (gh *GatewayHandler) SetRtHookExecutor(ex *runtime.RtHookExecutor)

SetRtHookExecutor configures BeforeRt/AfterRt for WebSocket envelopes.

func (*GatewayHandler) SetStatusRegistry

func (gh *GatewayHandler) SetStatusRegistry(sr *presence.StatusRegistry)

SetStatusRegistry configures the status follow graph and fan-out.

func (*GatewayHandler) SetStreamTracker

func (gh *GatewayHandler) SetStreamTracker(st *presence.StreamTracker)

SetStreamTracker configures the Local-first stream tracker (party/channel/match/status modes).

func (*GatewayHandler) StartRelayFanoutListener

func (gh *GatewayHandler) StartRelayFanoutListener(parent context.Context)

StartRelayFanoutListener subscribes to cross-node relayed match envelopes and delivers locally.

func (*GatewayHandler) StatusFollow

func (gh *GatewayHandler) StatusFollow(sessionID string, userIDs []string) error

StatusFollow adds follow edges for a session (runtime nk.status_follow).

func (*GatewayHandler) StatusUnfollow

func (gh *GatewayHandler) StatusUnfollow(sessionID string, userIDs []string) error

StatusUnfollow removes follow edges for a session (runtime nk.status_unfollow).

func (*GatewayHandler) TrackStatusOnConnect

func (gh *GatewayHandler) TrackStatusOnConnect(s *Session)

TrackStatusOnConnect auto-follows self and optionally Tracks status stream (?status=true).

func (*GatewayHandler) UntrackStatusOnDisconnect

func (gh *GatewayHandler) UntrackStatusOnDisconnect(sessionID, userID, username string)

UntrackStatusOnDisconnect removes status presence and follow edges after grace expiry.

func (*GatewayHandler) Upgrade

func (gh *GatewayHandler) Upgrade(w http.ResponseWriter, r *http.Request)

Upgrade upgrades HTTP requests to WebSocket connection stream.

type MatchCreatePayload

type MatchCreatePayload struct{}

type MatchDataSendPayload

type MatchDataSendPayload struct {
	MatchID   string   `json:"match_id"`
	OpCode    int64    `json:"op_code"`
	Data      string   `json:"data"` // Base64 or raw action payload
	Reliable  bool     `json:"reliable"`
	Presences []string `json:"presences,omitempty"` // session IDs
}

type MatchJoinPayload

type MatchJoinPayload struct {
	MatchID  string            `json:"match_id"`
	Token    string            `json:"token,omitempty"`
	Metadata map[string]string `json:"metadata,omitempty"`
}

type MatchLeavePayload

type MatchLeavePayload struct {
	MatchID string `json:"match_id"`
}

type MatchmakerAddPayload

type MatchmakerAddPayload struct {
	QueueName         string             `json:"queue_name"`
	MinCount          int                `json:"min_count"`
	MaxCount          int                `json:"max_count"`
	StringProperties  map[string]string  `json:"string_properties"`
	NumericProperties map[string]float64 `json:"numeric_properties"`
	CountMultiple     int                `json:"count_multiple"`
	ReversePrecision  bool               `json:"reverse_precision"`
	Query             string             `json:"query"`
}

type MatchmakerRemovePayload

type MatchmakerRemovePayload struct {
	TicketID string `json:"ticket_id"`
}

type PartyAcceptPayload

type PartyAcceptPayload struct {
	PartyID  string               `json:"party_id"`
	Presence PartyPresencePayload `json:"presence"`
}

type PartyClosePayload

type PartyClosePayload struct {
	PartyID string `json:"party_id"`
}

type PartyCreatePayload

type PartyCreatePayload struct {
	Open    bool   `json:"open"`
	Hidden  bool   `json:"hidden"`
	MaxSize int    `json:"max_size"`
	Label   string `json:"label"`
}

type PartyDataSendPayload

type PartyDataSendPayload struct {
	PartyID string `json:"party_id"`
	OpCode  int64  `json:"op_code"`
	Data    string `json:"data"`
}

type PartyJoinPayload

type PartyJoinPayload struct {
	PartyID string `json:"party_id"`
}

type PartyJoinRequestListPayload

type PartyJoinRequestListPayload struct {
	PartyID string `json:"party_id"`
}

type PartyLeavePayload

type PartyLeavePayload struct {
	PartyID string `json:"party_id"`
}

type PartyMatchmakerAddPayload

type PartyMatchmakerAddPayload struct {
	PartyID           string             `json:"party_id"`
	QueueName         string             `json:"queue_name"`
	MinCount          int                `json:"min_count"`
	MaxCount          int                `json:"max_count"`
	StringProperties  map[string]string  `json:"string_properties"`
	NumericProperties map[string]float64 `json:"numeric_properties"`
	CountMultiple     int                `json:"count_multiple"`
	Query             string             `json:"query"`
}

type PartyMatchmakerRemovePayload

type PartyMatchmakerRemovePayload struct {
	PartyID  string `json:"party_id"`
	TicketID string `json:"ticket_id"`
}

type PartyPresencePayload

type PartyPresencePayload struct {
	UserID    string `json:"user_id"`
	Username  string `json:"username"`
	SessionID string `json:"session_id"`
}

type PartyPromotePayload

type PartyPromotePayload struct {
	PartyID  string               `json:"party_id"`
	Presence PartyPresencePayload `json:"presence"`
}

type PartyRemovePayload

type PartyRemovePayload struct {
	PartyID  string               `json:"party_id"`
	Presence PartyPresencePayload `json:"presence"`
}

type PartyUpdatePayload

type PartyUpdatePayload struct {
	PartyID string `json:"party_id"`
	Open    bool   `json:"open"`
	Hidden  bool   `json:"hidden"`
	Label   string `json:"label"`
}

type PingPayload

type PingPayload struct{}

type RelayedMatch

type RelayedMatch struct {
	MatchID  string
	Sessions map[string]*Session
	// contains filtered or unexported fields
}

RelayedMatch represents a client-relayed room session.

type RpcPayload

type RpcPayload struct {
	ID      string `json:"id"`
	Payload string `json:"payload"`
}

type Session

type Session struct {
	ID          string
	UserID      string
	Username    string
	Conn        *websocket.Conn
	Send        chan []byte
	CloseOnce   sync.Once
	IsActive    bool
	TrackStatus bool
	// contains filtered or unexported fields
}

Session wraps a live WebSocket client connection.

func (*Session) CloseConn

func (s *Session) CloseConn()

CloseConn closes the underlying WebSocket connection (nil-safe).

func (*Session) MatchIDs

func (s *Session) MatchIDs() []string

MatchIDs returns a copy of joined match IDs.

func (*Session) TrySend

func (s *Session) TrySend(payload []byte)

TrySend enqueues a payload if the session is active; closes the connection under backpressure.

type StatusFollowPayload

type StatusFollowPayload struct {
	UserIDs   []string `json:"user_ids"`
	Usernames []string `json:"usernames"`
}

type StatusUnfollowPayload

type StatusUnfollowPayload struct {
	UserIDs []string `json:"user_ids"`
}

type StatusUpdatePayload

type StatusUpdatePayload struct {
	Status *string `json:"status"`
}

StatusUpdatePayload distinguishes JSON null (offline) from empty string (online blank).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL