store

package
v0.0.0-...-dfb2feb Latest Latest
Warning

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

 Go to latest Published: Jan 22, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultVirtualNodes = 150

DefaultVirtualNodes is the default number of virtual nodes per server for consistent hashing. Higher values provide better distribution.

Variables

View Source 
var (
	ErrNoServers       = errors.New("no servers available")
	ErrServerNotFound  = errors.New("server not found")
	ErrServerExists    = errors.New("server already exists")
	ErrRoomNotAssigned = errors.New("room not assigned to any server")
)

Coordinator errors.

View Source 
var (
	// ErrSessionNotFound is returned when a session is not found.
	ErrSessionNotFound = errors.New("session not found")
	// ErrSessionExpired is returned when a session has expired.
	ErrSessionExpired = errors.New("session expired")
)
View Source 
var (
	ErrRoomNotFound      = errors.New("room not found")
	ErrRoomAlreadyExists = errors.New("room already exists")
	ErrInvalidRoomInfo   = errors.New("invalid room info")
)

RoomStore errors.

Functions

This section is empty.

Types

type GoRedisClient 

type GoRedisClient struct {
	// contains filtered or unexported fields
}

GoRedisClient is an adapter for go-redis/v9 that implements the RedisClient interface.

func NewGoRedisClient 

func NewGoRedisClient(opts *redis.Options) (*GoRedisClient, error)

NewGoRedisClient creates a new GoRedisClient with the provided options.

func (*GoRedisClient) Close 

func (c *GoRedisClient) Close() error

Close closes the Redis client connection.

func (*GoRedisClient) Del 

func (c *GoRedisClient) Del(ctx context.Context, keys ...string) error

Del deletes one or more keys.

func (*GoRedisClient) Exists 

func (c *GoRedisClient) Exists(ctx context.Context, keys ...string) (int64, error)

Exists checks if keys exist.

func (*GoRedisClient) Expire 

func (c *GoRedisClient) Expire(ctx context.Context, key string, expiration time.Duration) error

Expire sets a timeout on a key.

func (*GoRedisClient) Get 

func (c *GoRedisClient) Get(ctx context.Context, key string) (string, error)

Get retrieves a value by key.

func (*GoRedisClient) HGet 

func (c *GoRedisClient) HGet(ctx context.Context, key, field string) (string, error)

HGet retrieves a hash field value.

func (*GoRedisClient) HGetAll 

func (c *GoRedisClient) HGetAll(ctx context.Context, key string) (map[string]string, error)

HGetAll retrieves all hash fields and values.

func (*GoRedisClient) HIncrBy 

func (c *GoRedisClient) HIncrBy(ctx context.Context, key, field string, incr int64) (int64, error)

HIncrBy increments a hash field by an integer.

func (*GoRedisClient) HSet 

func (c *GoRedisClient) HSet(ctx context.Context, key string, values ...interface{}) error

HSet sets hash field-value pairs.

func (*GoRedisClient) Keys 

func (c *GoRedisClient) Keys(ctx context.Context, pattern string) ([]string, error)

Keys returns all keys matching a pattern.

func (*GoRedisClient) SAdd 

func (c *GoRedisClient) SAdd(ctx context.Context, key string, members ...interface{}) error

SAdd adds one or more members to a set.

func (*GoRedisClient) SMembers 

func (c *GoRedisClient) SMembers(ctx context.Context, key string) ([]string, error)

SMembers retrieves all members of a set.

func (*GoRedisClient) SRem 

func (c *GoRedisClient) SRem(ctx context.Context, key string, members ...interface{}) error

SRem removes one or more members from a set.

func (*GoRedisClient) Set 

func (c *GoRedisClient) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) error

Set sets a key-value pair with an expiration time.

type HashRingCoordinator 

type HashRingCoordinator struct {
	// contains filtered or unexported fields
}

HashRingCoordinator implements RoomCoordinator using consistent hashing. It distributes rooms across SFU servers with minimal remapping when servers are added or removed.

func NewHashRingCoordinator 

func NewHashRingCoordinator(opts ...HashRingOption) *HashRingCoordinator

NewHashRingCoordinator creates a new consistent hash ring coordinator.

func (*HashRingCoordinator) AddServer 

func (c *HashRingCoordinator) AddServer(serverID string) error

AddServer adds a server to the consistent hash ring.

func (*HashRingCoordinator) AssignRoom 

func (c *HashRingCoordinator) AssignRoom(ctx context.Context, roomID string) (string, error)

AssignRoom assigns a room to an SFU server using consistent hashing. If the room is already assigned, it returns the existing assignment.

func (*HashRingCoordinator) GetDistribution 

func (c *HashRingCoordinator) GetDistribution() map[string]int

GetDistribution returns the distribution of virtual nodes per server. Useful for debugging and monitoring.

func (*HashRingCoordinator) GetServerForRoom 

func (c *HashRingCoordinator) GetServerForRoom(ctx context.Context, roomID string) (string, error)

GetServerForRoom returns the assigned server for a room. It first checks the room store, then falls back to consistent hashing.

func (*HashRingCoordinator) GetServers 

func (c *HashRingCoordinator) GetServers() []string

GetServers returns all servers in the ring.

func (*HashRingCoordinator) RebalanceRoom 

func (c *HashRingCoordinator) RebalanceRoom(ctx context.Context, roomID, newServerID string) error

RebalanceRoom moves a room to a different server for failover.

func (*HashRingCoordinator) RemoveServer 

func (c *HashRingCoordinator) RemoveServer(serverID string) error

RemoveServer removes a server from the consistent hash ring.

func (*HashRingCoordinator) ServerCount 

func (c *HashRingCoordinator) ServerCount() int

ServerCount returns the number of servers in the ring.

type HashRingOption 

type HashRingOption func(*HashRingCoordinator)

HashRingOption is a functional option for configuring HashRingCoordinator.

func WithRoomStore 

func WithRoomStore(store RoomStore) HashRingOption

WithRoomStore sets the room store for persisting room assignments.

func WithVirtualNodes 

func WithVirtualNodes(n int) HashRingOption

WithVirtualNodes sets the number of virtual nodes per server.

type MemoryRoomStore 

type MemoryRoomStore struct {
	// contains filtered or unexported fields
}

MemoryRoomStore is an in-memory implementation of RoomStore. Useful for testing and single-instance deployments.

func NewMemoryRoomStore 

func NewMemoryRoomStore() *MemoryRoomStore

NewMemoryRoomStore creates a new in-memory room store.

func (*MemoryRoomStore) Close 

func (s *MemoryRoomStore) Close() error

Close cleans up resources.

func (*MemoryRoomStore) DeleteRoom 

func (s *MemoryRoomStore) DeleteRoom(_ context.Context, roomID string) error

DeleteRoom deletes room information from memory.

func (*MemoryRoomStore) GetRoom 

func (s *MemoryRoomStore) GetRoom(_ context.Context, roomID string) (*RoomInfo, error)

GetRoom retrieves room information by room ID.

func (*MemoryRoomStore) ListRooms 

func (s *MemoryRoomStore) ListRooms(_ context.Context) ([]*RoomInfo, error)

ListRooms returns all rooms in the store.

func (*MemoryRoomStore) ListRoomsByServer 

func (s *MemoryRoomStore) ListRoomsByServer(_ context.Context, serverID string) ([]*RoomInfo, error)

ListRoomsByServer returns all rooms assigned to a specific server.

func (*MemoryRoomStore) RoomExists 

func (s *MemoryRoomStore) RoomExists(_ context.Context, roomID string) (bool, error)

RoomExists checks if a room exists in the store.

func (*MemoryRoomStore) SaveRoom 

func (s *MemoryRoomStore) SaveRoom(_ context.Context, room *RoomInfo) error

SaveRoom saves room information to memory.

func (*MemoryRoomStore) UpdateParticipantCount 

func (s *MemoryRoomStore) UpdateParticipantCount(_ context.Context, roomID string, delta int) error

UpdateParticipantCount atomically updates the participant count for a room.

func (*MemoryRoomStore) UpdateRoom 

func (s *MemoryRoomStore) UpdateRoom(_ context.Context, room *RoomInfo) error

UpdateRoom updates existing room information in memory.

func (*MemoryRoomStore) UpdateRoomState 

func (s *MemoryRoomStore) UpdateRoomState(_ context.Context, roomID string, state RoomState) error

UpdateRoomState updates the state of a room.

func (*MemoryRoomStore) UpdateTrackCount 

func (s *MemoryRoomStore) UpdateTrackCount(_ context.Context, roomID string, delta int) error

UpdateTrackCount atomically updates the track count for a room.

type MemoryStore 

type MemoryStore struct {
	// contains filtered or unexported fields
}

MemoryStore is an in-memory implementation of SessionStore.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates a new in-memory session store.

func (*MemoryStore) Close 

func (s *MemoryStore) Close() error

Close closes the store and cleans up resources.

func (*MemoryStore) DeleteSession 

func (s *MemoryStore) DeleteSession(ctx context.Context, sessionID string) error

DeleteSession deletes a session from the store.

func (*MemoryStore) GetSession 

func (s *MemoryStore) GetSession(ctx context.Context, sessionID string) (*Session, error)

GetSession retrieves a session from the store by session ID.

func (*MemoryStore) GetSessionsByParticipant 

func (s *MemoryStore) GetSessionsByParticipant(ctx context.Context, participantID string) ([]*Session, error)

GetSessionsByParticipant retrieves all sessions for a given participant.

func (*MemoryStore) GetSessionsByRoom 

func (s *MemoryStore) GetSessionsByRoom(ctx context.Context, roomID string) ([]*Session, error)

GetSessionsByRoom retrieves all sessions for a given room.

func (*MemoryStore) SaveSession 

func (s *MemoryStore) SaveSession(ctx context.Context, session *Session) error

SaveSession saves a session to the store.

func (*MemoryStore) UpdateSession 

func (s *MemoryStore) UpdateSession(ctx context.Context, session *Session) error

UpdateSession updates an existing session in the store.

type RedisClient 

type RedisClient interface {
	Set(ctx context.Context, key string, value interface{}, expiration time.Duration) error
	Get(ctx context.Context, key string) (string, error)
	Del(ctx context.Context, keys ...string) error
	SAdd(ctx context.Context, key string, members ...interface{}) error
	SMembers(ctx context.Context, key string) ([]string, error)
	SRem(ctx context.Context, key string, members ...interface{}) error
	Expire(ctx context.Context, key string, expiration time.Duration) error
	Exists(ctx context.Context, keys ...string) (int64, error)
	HSet(ctx context.Context, key string, values ...interface{}) error
	HGet(ctx context.Context, key string, field string) (string, error)
	HGetAll(ctx context.Context, key string) (map[string]string, error)
	HIncrBy(ctx context.Context, key string, field string, incr int64) (int64, error)
	Keys(ctx context.Context, pattern string) ([]string, error)
	Close() error
}

RedisClient is an interface for Redis operations. This allows for easier testing and mocking.

type RedisRoomStore 

type RedisRoomStore struct {
	// contains filtered or unexported fields
}

RedisRoomStore is a Redis-based implementation of RoomStore. It uses Redis Hash for storing room information (ADR-0005). Unlike session store, rooms do not have TTL and must be explicitly deleted.

func NewRedisRoomStore 

func NewRedisRoomStore(client RedisClient, prefix string) *RedisRoomStore

NewRedisRoomStore creates a new Redis room store.

func (*RedisRoomStore) Close 

func (s *RedisRoomStore) Close() error

Close closes the Redis client.

func (*RedisRoomStore) DeleteRoom 

func (s *RedisRoomStore) DeleteRoom(ctx context.Context, roomID string) error

DeleteRoom deletes room information from Redis.

func (*RedisRoomStore) GetRoom 

func (s *RedisRoomStore) GetRoom(ctx context.Context, roomID string) (*RoomInfo, error)

GetRoom retrieves room information from Redis by room ID.

func (*RedisRoomStore) ListRooms 

func (s *RedisRoomStore) ListRooms(ctx context.Context) ([]*RoomInfo, error)

ListRooms returns all rooms in the store.

func (*RedisRoomStore) ListRoomsByServer 

func (s *RedisRoomStore) ListRoomsByServer(ctx context.Context, serverID string) ([]*RoomInfo, error)

ListRoomsByServer returns all rooms assigned to a specific server.

func (*RedisRoomStore) RoomExists 

func (s *RedisRoomStore) RoomExists(ctx context.Context, roomID string) (bool, error)

RoomExists checks if a room exists in the store.

func (*RedisRoomStore) SaveRoom 

func (s *RedisRoomStore) SaveRoom(ctx context.Context, room *RoomInfo) error

SaveRoom saves room information to Redis. Rooms do not have TTL and must be explicitly deleted (ADR-0005).

func (*RedisRoomStore) UpdateParticipantCount 

func (s *RedisRoomStore) UpdateParticipantCount(ctx context.Context, roomID string, delta int) error

UpdateParticipantCount atomically updates the participant count for a room.

func (*RedisRoomStore) UpdateRoom 

func (s *RedisRoomStore) UpdateRoom(ctx context.Context, room *RoomInfo) error

UpdateRoom updates existing room information in Redis.

func (*RedisRoomStore) UpdateRoomState 

func (s *RedisRoomStore) UpdateRoomState(ctx context.Context, roomID string, state RoomState) error

UpdateRoomState updates the state of a room.

func (*RedisRoomStore) UpdateTrackCount 

func (s *RedisRoomStore) UpdateTrackCount(ctx context.Context, roomID string, delta int) error

UpdateTrackCount atomically updates the track count for a room.

type RedisStore 

type RedisStore struct {
	// contains filtered or unexported fields
}

RedisStore is a Redis-based implementation of SessionStore.

func NewRedisStore 

func NewRedisStore(client RedisClient, prefix string) *RedisStore

NewRedisStore creates a new Redis session store.

func (*RedisStore) Close 

func (s *RedisStore) Close() error

Close closes the Redis client.

func (*RedisStore) DeleteSession 

func (s *RedisStore) DeleteSession(ctx context.Context, sessionID string) error

DeleteSession deletes a session from Redis.

func (*RedisStore) GetSession 

func (s *RedisStore) GetSession(ctx context.Context, sessionID string) (*Session, error)

GetSession retrieves a session from Redis by session ID.

func (*RedisStore) GetSessionsByParticipant 

func (s *RedisStore) GetSessionsByParticipant(ctx context.Context, participantID string) ([]*Session, error)

GetSessionsByParticipant retrieves all sessions for a given participant.

func (*RedisStore) GetSessionsByRoom 

func (s *RedisStore) GetSessionsByRoom(ctx context.Context, roomID string) ([]*Session, error)

GetSessionsByRoom retrieves all sessions for a given room.

func (*RedisStore) SaveSession 

func (s *RedisStore) SaveSession(ctx context.Context, session *Session) error

SaveSession saves a session to Redis.

func (*RedisStore) UpdateSession 

func (s *RedisStore) UpdateSession(ctx context.Context, session *Session) error

UpdateSession updates an existing session in Redis.

type RoomCoordinator 

type RoomCoordinator interface {
	// AssignRoom assigns a room to an SFU server using consistent hashing.
	AssignRoom(ctx context.Context, roomID string) (serverID string, err error)

	// GetServerForRoom returns the assigned server for a room.
	GetServerForRoom(ctx context.Context, roomID string) (serverID string, err error)

	// RebalanceRoom moves a room to a different server for failover.
	RebalanceRoom(ctx context.Context, roomID string, newServerID string) error

	// AddServer adds a server to the consistent hash ring.
	AddServer(serverID string) error

	// RemoveServer removes a server from the consistent hash ring.
	RemoveServer(serverID string) error

	// GetServers returns all servers in the ring.
	GetServers() []string
}

RoomCoordinator manages room-to-server assignments using consistent hashing.

type RoomInfo 

type RoomInfo struct {
	// Metadata is arbitrary metadata associated with the room.
	Metadata map[string]interface{} `json:"metadata,omitempty"`
	// CreatedAt is the time when the room was created.
	CreatedAt time.Time `json:"created_at"`
	// UpdatedAt is the time when the room was last updated.
	UpdatedAt time.Time `json:"updated_at"`
	// RoomID is the unique identifier for the room.
	RoomID string `json:"room_id"`
	// ServerID is the ID of the SFU server handling this room.
	ServerID string `json:"server_id"`
	// State is the current state of the room.
	State RoomState `json:"state"`
	// ParticipantCount is the number of participants in the room.
	ParticipantCount int `json:"participant_count"`
	// MaxParticipants is the maximum number of participants allowed.
	MaxParticipants int `json:"max_participants"`
	// TrackCount is the current number of tracks in the room.
	TrackCount int `json:"track_count"`
}

RoomInfo represents room information stored in the distributed store. This is used for room-to-server mapping in a multi-instance SFU deployment.

type RoomState 

type RoomState string

RoomState represents the state of a room in the distributed store. 

const (
	// RoomStateCreated indicates the room has been created but has no participants.
	RoomStateCreated RoomState = "created"
	// RoomStateActive indicates the room has one or more participants.
	RoomStateActive RoomState = "active"
	// RoomStateLocked indicates the room is locked and new participants cannot join.
	RoomStateLocked RoomState = "locked"
	// RoomStateClosing indicates the room is in the process of closing.
	RoomStateClosing RoomState = "closing"
	// RoomStateClosed indicates the room has been closed.
	RoomStateClosed RoomState = "closed"
)

type RoomStore 

type RoomStore interface {
	// SaveRoom saves room information to the store.
	// Unlike sessions, rooms do not have TTL and must be explicitly deleted (ADR-0005).
	SaveRoom(ctx context.Context, room *RoomInfo) error

	// GetRoom retrieves room information by room ID.
	GetRoom(ctx context.Context, roomID string) (*RoomInfo, error)

	// DeleteRoom deletes room information from the store.
	DeleteRoom(ctx context.Context, roomID string) error

	// UpdateRoom updates existing room information in the store.
	UpdateRoom(ctx context.Context, room *RoomInfo) error

	// ListRooms returns all rooms in the store.
	ListRooms(ctx context.Context) ([]*RoomInfo, error)

	// ListRoomsByServer returns all rooms assigned to a specific server.
	ListRoomsByServer(ctx context.Context, serverID string) ([]*RoomInfo, error)

	// RoomExists checks if a room exists in the store.
	RoomExists(ctx context.Context, roomID string) (bool, error)

	// UpdateParticipantCount atomically updates the participant count for a room.
	UpdateParticipantCount(ctx context.Context, roomID string, delta int) error

	// UpdateTrackCount atomically updates the track count for a room.
	UpdateTrackCount(ctx context.Context, roomID string, delta int) error

	// UpdateRoomState updates the state of a room.
	UpdateRoomState(ctx context.Context, roomID string, state RoomState) error

	// Close closes the store and cleans up resources.
	Close() error
}

RoomStore is an interface for distributed room registry. It manages room information across multiple SFU instances.

type Session 

type Session struct {
	// SessionID is the unique identifier for the session.
	SessionID string
	// ParticipantID is the ID of the participant associated with this session.
	ParticipantID string
	// RoomID is the ID of the room the participant is in.
	RoomID string
	// PublishedTracks is a list of track IDs published by this participant.
	PublishedTracks []string
	// Subscriptions is a list of subscription IDs for this participant.
	Subscriptions []string
	// Metadata is arbitrary metadata associated with the session.
	Metadata map[string]interface{}
	// UserAgent is the user agent string of the client.
	UserAgent string
	// IPAddress is the IP address of the client.
	IPAddress string
	// CreatedAt is the time when the session was created.
	CreatedAt time.Time
	// ExpiresAt is the time when the session expires.
	ExpiresAt time.Time
}

Session represents a session with its associated data.

type SessionStore 

type SessionStore interface {
	// SaveSession saves a session to the store.
	SaveSession(ctx context.Context, session *Session) error

	// GetSession retrieves a session from the store by session ID.
	GetSession(ctx context.Context, sessionID string) (*Session, error)

	// DeleteSession deletes a session from the store.
	DeleteSession(ctx context.Context, sessionID string) error

	// UpdateSession updates an existing session in the store.
	UpdateSession(ctx context.Context, session *Session) error

	// GetSessionsByParticipant retrieves all sessions for a given participant.
	GetSessionsByParticipant(ctx context.Context, participantID string) ([]*Session, error)

	// GetSessionsByRoom retrieves all sessions for a given room.
	GetSessionsByRoom(ctx context.Context, roomID string) ([]*Session, error)

	// Close closes the store and cleans up resources.
	Close() error
}

SessionStore is an interface for storing and retrieving session data.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.