mk6i-open-oscar-server/state/session_manager.go

package state

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"github.com/mk6i/open-oscar-server/wire"
)

type sessionSlot struct {
	session      *Session
	removed      chan bool
	multiSession bool
}

type userLock struct {
	sync.Mutex
	refCount int
}

// InMemorySessionManager handles the lifecycle of a user session and provides
// synchronized message relay between sessions in the session pool. An
// InMemorySessionManager is safe for concurrent use by multiple goroutines.
type InMemorySessionManager struct {
	store                 map[IdentScreenName]*sessionSlot
	mapMutex              sync.RWMutex
	userLocks             map[IdentScreenName]*userLock
	userLocksMutex        sync.Mutex
	logger                *slog.Logger
	maxConcurrentSessions int
}

const (
	// DefaultMaxConcurrentSessions is the default maximum number of concurrent
	// sessions allowed when multi-session is enabled.
	DefaultMaxConcurrentSessions = 5
)

// ErrMaxConcurrentSessionsReached is returned when attempting to add a new
// session instance but the maximum number of concurrent sessions has been
// reached.
var ErrMaxConcurrentSessionsReached = errors.New("maximum number of concurrent sessions reached")

// NewInMemorySessionManager creates a new instance of InMemorySessionManager.
func NewInMemorySessionManager(logger *slog.Logger) *InMemorySessionManager {
	return &InMemorySessionManager{
		logger:                logger,
		store:                 make(map[IdentScreenName]*sessionSlot),
		userLocks:             make(map[IdentScreenName]*userLock),
		maxConcurrentSessions: DefaultMaxConcurrentSessions,
	}
}

func (s *InMemorySessionManager) lockUser(sn IdentScreenName) {
	s.userLocksMutex.Lock()

	lock, ok := s.userLocks[sn]
	if !ok {
		lock = &userLock{}
		s.userLocks[sn] = lock
	}

	lock.refCount++
	s.userLocksMutex.Unlock()

	lock.Lock()
}

func (s *InMemorySessionManager) unlockUser(sn IdentScreenName) {
	s.userLocksMutex.Lock()
	defer s.userLocksMutex.Unlock()

	lock, ok := s.userLocks[sn]
	if !ok {
		return
	}

	lock.Unlock()
	lock.refCount--

	if lock.refCount == 0 {
		delete(s.userLocks, sn)
	}
}

// RelayToAll relays a message to all sessions in the session pool.
func (s *InMemorySessionManager) RelayToAll(ctx context.Context, msg wire.SNACMessage) {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()
	for _, rec := range s.store {
		s.maybeRelayMessage(ctx, msg, rec.session)
	}
}

// RelayToScreenName relays a message to a session with a matching screen name.
func (s *InMemorySessionManager) RelayToScreenName(ctx context.Context, screenName IdentScreenName, msg wire.SNACMessage) {
	sess := s.RetrieveSession(screenName)
	if sess == nil {
		s.logger.WarnContext(ctx, "RelayToScreenName: session not found",
			"recipient", screenName,
			"food_group", wire.FoodGroupName(msg.Frame.FoodGroup),
			"sub_group", wire.SubGroupName(msg.Frame.FoodGroup, msg.Frame.SubGroup))
		return
	}
	s.logger.DebugContext(ctx, "RelayToScreenName: found session, relaying", "recipient", screenName,
		"food_group", wire.FoodGroupName(msg.Frame.FoodGroup),
		"sub_group", wire.SubGroupName(msg.Frame.FoodGroup, msg.Frame.SubGroup),
		"instances", len(sess.Instances()))
	s.maybeRelayMessage(ctx, msg, sess)
}

// RelayToScreenNames relays a message to sessions with matching screenNames.
func (s *InMemorySessionManager) RelayToScreenNames(ctx context.Context, screenNames []IdentScreenName, msg wire.SNACMessage) {
	for _, sess := range s.retrieveByScreenNames(screenNames) {
		s.maybeRelayMessage(ctx, msg, sess)
	}
}

func (s *InMemorySessionManager) RelayToSelf(ctx context.Context, instance *SessionInstance, msg wire.SNACMessage) {
	switch instance.RelayMessageToInstance(msg) {
	case SessSendClosed:
		s.logger.WarnContext(ctx, "can't send notification because the user's session is closed", "recipient", instance.IdentScreenName(), "message", msg)
	case SessQueueFull:
		s.logger.WarnContext(ctx, "can't send notification because queue is full", "recipient", instance.IdentScreenName(), "message", msg)
		instance.CloseInstance()
	}
}

func (s *InMemorySessionManager) RelayToOtherInstances(ctx context.Context, instance *SessionInstance, msg wire.SNACMessage) {
	for _, inst := range instance.Session().Instances() {
		if instance == inst || !inst.live() {
			continue
		}
		switch inst.RelayMessageToInstance(msg) {
		case SessSendClosed:
			s.logger.WarnContext(ctx, "can't send notification because the user's session is closed", "recipient", instance.IdentScreenName(), "message", msg)
		case SessQueueFull:
			s.logger.WarnContext(ctx, "can't send notification because queue is full", "recipient", instance.IdentScreenName(), "message", msg)
			inst.CloseInstance()
		}
	}
}

func (s *InMemorySessionManager) RelayToScreenNameActiveOnly(ctx context.Context, screenName IdentScreenName, msg wire.SNACMessage) {
	sess := s.RetrieveSession(screenName)
	if sess == nil {
		s.logger.WarnContext(ctx, "RelayToScreenNameActiveOnly: session not found",
			"recipient", screenName,
			"food_group", wire.FoodGroupName(msg.Frame.FoodGroup),
			"sub_group", wire.SubGroupName(msg.Frame.FoodGroup, msg.Frame.SubGroup))
		return
	}
	s.logger.DebugContext(ctx, "RelayToScreenNameActiveOnly: found session, relaying", "recipient", screenName,
		"food_group", wire.FoodGroupName(msg.Frame.FoodGroup),
		"sub_group", wire.SubGroupName(msg.Frame.FoodGroup, msg.Frame.SubGroup),
		"instances", len(sess.Instances()),
		"inactive", sess.Inactive())
	s.maybeRelayMessageActiveOnly(ctx, msg, sess)
}

func (s *InMemorySessionManager) maybeRelayMessage(ctx context.Context, msg wire.SNACMessage, sess *Session) {
	for _, instance := range sess.Instances() {
		if !instance.live() {
			s.logger.DebugContext(ctx, "maybeRelayMessage: skipping non-live instance",
				"recipient", sess.IdentScreenName(),
				"food_group", wire.FoodGroupName(msg.Frame.FoodGroup),
				"sub_group", wire.SubGroupName(msg.Frame.FoodGroup, msg.Frame.SubGroup))
			continue
		}
		switch instance.RelayMessageToInstance(msg) {
		case SessSendClosed:
			s.logger.WarnContext(ctx, "can't send notification because the user's session is closed", "recipient", sess.IdentScreenName(), "message", msg)
		case SessQueueFull:
			s.logger.WarnContext(ctx, "can't send notification because queue is full", "recipient", sess.IdentScreenName(), "message", msg)
			instance.CloseInstance()
		default:
			s.logger.DebugContext(ctx, "maybeRelayMessage: relayed to instance",
				"recipient", sess.IdentScreenName(),
				"food_group", wire.FoodGroupName(msg.Frame.FoodGroup),
				"sub_group", wire.SubGroupName(msg.Frame.FoodGroup, msg.Frame.SubGroup),
			)
		}
	}
}

func (s *InMemorySessionManager) maybeRelayMessageActiveOnly(ctx context.Context, msg wire.SNACMessage, sess *Session) {
	for _, instance := range sess.Instances() {
		if !instance.active() {
			continue
		}
		switch instance.RelayMessageToInstance(msg) {
		case SessSendClosed:
			s.logger.WarnContext(ctx, "can't send notification because the user's session is closed", "recipient", sess.IdentScreenName(), "message", msg)
		case SessQueueFull:
			s.logger.WarnContext(ctx, "can't send notification because queue is full", "recipient", sess.IdentScreenName(), "message", msg)
			instance.CloseInstance()
		}
	}
}

func (s *InMemorySessionManager) AddSession(ctx context.Context, screenName DisplayScreenName, doMultiSess bool, cfg ...func(sess *Session)) (*SessionInstance, error) {
	s.lockUser(screenName.IdentScreenName())
	defer s.unlockUser(screenName.IdentScreenName())

	s.mapMutex.Lock()
	active := s.findRec(screenName.IdentScreenName())
	s.mapMutex.Unlock()

	if active != nil {
		if doMultiSess {
			if !active.multiSession {
				active.session.CloseSession()
				return s.newSession(screenName, doMultiSess, cfg)
			}

			// Check if we've reached the maximum number of concurrent sessions
			if active.session.InstanceCount() >= s.maxConcurrentSessions {
				return nil, fmt.Errorf("%w: max instance(s) = %d", ErrMaxConcurrentSessionsReached, s.maxConcurrentSessions)
			}

			instance := active.session.AddInstance()

			return instance, nil
		} else {
			// signal to callers that this session group has to go
			active.session.CloseSession()

			select {
			case <-active.removed: // wait for RemoveSession to be called
			case <-ctx.Done():
				return nil, fmt.Errorf("waiting for previous session to terminate: %w", ctx.Err())
			}
		}
	}

	return s.newSession(screenName, doMultiSess, cfg)
}

func (s *InMemorySessionManager) newSession(screenName DisplayScreenName, doMultiSess bool, cfg []func(sess *Session)) (*SessionInstance, error) {
	sess := NewSession()
	sess.SetIdentScreenName(screenName.IdentScreenName())
	sess.SetDisplayScreenName(screenName)

	for _, f := range cfg {
		f(sess)
	}

	// Create a new instance within the session group
	instance := sess.AddInstance()

	s.mapMutex.Lock()
	s.store[instance.IdentScreenName()] = &sessionSlot{
		session:      sess,
		removed:      make(chan bool),
		multiSession: doMultiSess,
	}
	s.mapMutex.Unlock()

	return instance, nil
}

func (s *InMemorySessionManager) findRec(identScreenName IdentScreenName) *sessionSlot {
	for _, rec := range s.store {
		if identScreenName == rec.session.IdentScreenName() {
			return rec
		}
	}
	return nil
}

// RemoveSession takes a session out of the session pool.
func (s *InMemorySessionManager) RemoveSession(session *Session) {
	s.mapMutex.Lock()
	defer s.mapMutex.Unlock()
	if rec, ok := s.store[session.IdentScreenName()]; ok && rec.session == session {
		delete(s.store, session.IdentScreenName())
		close(rec.removed)
	}
}

// RetrieveSession finds a session with a matching screen name. Returns nil if
// session is not found or if there are no active instances with complete signon.
func (s *InMemorySessionManager) RetrieveSession(screenName IdentScreenName) *Session {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()
	if rec, ok := s.store[screenName]; ok {
		if rec.session.HasLiveInstances() {
			return rec.session
		}
	}
	return nil
}

func (s *InMemorySessionManager) retrieveByScreenNames(screenNames []IdentScreenName) []*Session {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()
	var ret []*Session
	for _, sn := range screenNames {
		for _, rec := range s.store {
			if sn == rec.session.IdentScreenName() {
				ret = append(ret, rec.session)
				break
			}
		}
	}
	return ret
}

// Empty returns true if the session pool contains 0 sessions.
func (s *InMemorySessionManager) Empty() bool {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()
	return len(s.store) == 0
}

// AllSessions returns all sessions in the session pool.
func (s *InMemorySessionManager) AllSessions() []*Session {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()
	var sessions []*Session
	for _, rec := range s.store {
		if !rec.session.HasLiveInstances() {
			continue
		}
		sessions = append(sessions, rec.session)
	}
	return sessions
}

// NewInMemoryChatSessionManager creates a new instance of
// InMemoryChatSessionManager.
func NewInMemoryChatSessionManager(logger *slog.Logger) *InMemoryChatSessionManager {
	return &InMemoryChatSessionManager{
		store:  make(map[string]*InMemorySessionManager),
		logger: logger,
	}
}

// InMemoryChatSessionManager manages chat sessions for multiple chat rooms
// stored in memory. It provides thread-safe operations to add, remove, and
// manipulate sessions as well as relay messages to participants.
type InMemoryChatSessionManager struct {
	logger   *slog.Logger
	mapMutex sync.RWMutex
	store    map[string]*InMemorySessionManager
}

// AddSession adds a user to a chat room. If screenName already exists, the old
// session is replaced by a new one. Optional cfg callbacks let callers attach
// setup (for example shutdown behavior) while the session is still being
// created.
func (s *InMemoryChatSessionManager) AddSession(ctx context.Context, chatCookie string, screenName DisplayScreenName, cfg ...func(sess *Session)) (*SessionInstance, error) {
	s.mapMutex.Lock()
	if _, ok := s.store[chatCookie]; !ok {
		s.store[chatCookie] = NewInMemorySessionManager(s.logger)
	}
	sessionManager := s.store[chatCookie]
	s.mapMutex.Unlock()

	ctx, cancel := context.WithTimeout(ctx, time.Second*5)
	defer cancel()

	sess, err := sessionManager.AddSession(ctx, screenName, false, cfg...)
	if err != nil {
		return nil, fmt.Errorf("AddSession: %w", err)
	}

	sess.Session().SetChatRoomCookie(chatCookie)

	s.mapMutex.Lock()
	defer s.mapMutex.Unlock()

	// at this point it's guaranteed that the prior chat session and corresponding
	// session manager (if the room count dropped to 0) were removed.
	//
	// - SessionManager.RemoveSession() was called because that unlocks
	//   SessionManager.AddSession(), which unblocks ChatSessionManager.AddSession()
	// - ChatSessionManager.RemoveSession() must call room deletion routine before
	//   releasing mapMutex
	//
	// now restore the chat session manager, which may have been deleted by the
	// call to RemoveSession().
	if _, ok := s.store[chatCookie]; !ok {
		s.store[chatCookie] = sessionManager
	}

	return sess, nil
}

// RemoveSession removes a user session from a chat room. It panics if you
// attempt to remove the session twice.
func (s *InMemoryChatSessionManager) RemoveSession(sess *Session) {
	s.mapMutex.Lock()
	defer s.mapMutex.Unlock()

	sessionManager, ok := s.store[sess.ChatRoomCookie()]
	if !ok {
		panic("attempting to remove a session after its room has been deleted")
	}
	sessionManager.RemoveSession(sess)

	if sessionManager.Empty() {
		delete(s.store, sess.ChatRoomCookie())
	}
}

// RemoveUserFromAllChats removes a user's session from all chat rooms.
func (s *InMemoryChatSessionManager) RemoveUserFromAllChats(user IdentScreenName) {
	var cpy []*InMemorySessionManager

	// make a copy since CloseSession() may call back to InMemoryChatSessionManager
	// and deadlock with this call
	s.mapMutex.RLock()
	for _, sessionManager := range s.store {
		cpy = append(cpy, sessionManager)
	}
	s.mapMutex.RUnlock()

	for _, sessionManager := range cpy {
		userSess := sessionManager.RetrieveSession(user)
		if userSess != nil {
			userSess.CloseSession()
		}
	}
}

// AllSessions returns all chat room participants. Returns
// ErrChatRoomNotFound if the room does not exist.
func (s *InMemoryChatSessionManager) AllSessions(cookie string) []*Session {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()

	sessionManager, ok := s.store[cookie]
	if !ok {
		s.logger.Debug("trying to get sessions for non-existent room", "cookie", cookie)
		return nil
	}
	return sessionManager.AllSessions()
}

// RelayToAllExcept sends a message to all chat room participants except for
// the participant with a particular screen name. Returns ErrChatRoomNotFound
// if the room does not exist for cookie.
func (s *InMemoryChatSessionManager) RelayToAllExcept(ctx context.Context, cookie string, except IdentScreenName, msg wire.SNACMessage) {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()

	sessionManager, ok := s.store[cookie]
	if !ok {
		s.logger.Error("trying to relay message to all for non-existent room", "cookie", cookie)
		return
	}

	for _, sess := range sessionManager.AllSessions() {
		if sess.IdentScreenName() == except {
			continue
		}
		sessionManager.maybeRelayMessage(ctx, msg, sess)
	}
}

// RelayToScreenName sends a message to a chat room user. Returns
// ErrChatRoomNotFound if the room does not exist for cookie.
func (s *InMemoryChatSessionManager) RelayToScreenName(ctx context.Context, cookie string, recipient IdentScreenName, msg wire.SNACMessage) {
	s.mapMutex.RLock()
	defer s.mapMutex.RUnlock()

	sessionManager, ok := s.store[cookie]
	if !ok {
		s.logger.Error("trying to relay message to screen name for non-existent room", "cookie", cookie)
		return
	}
	sessionManager.RelayToScreenName(ctx, recipient, msg)
}