session_manager.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162
  1. package state
  2. import (
  3. "context"
  4. "log/slog"
  5. "sync"
  6. "github.com/mk6i/retro-aim-server/wire"
  7. )
  8. // InMemorySessionManager handles the lifecycle of a user session and provides
  9. // synchronized message relay between sessions in the session pool. An
  10. // InMemorySessionManager is safe for concurrent use by multiple goroutines.
  11. type InMemorySessionManager struct {
  12. store map[string]*Session
  13. mapMutex sync.RWMutex
  14. logger *slog.Logger
  15. }
  16. // NewInMemorySessionManager creates a new instance of InMemorySessionManager.
  17. func NewInMemorySessionManager(logger *slog.Logger) *InMemorySessionManager {
  18. return &InMemorySessionManager{
  19. logger: logger,
  20. store: make(map[string]*Session),
  21. }
  22. }
  23. // RelayToAll relays a message to all sessions in the session pool.
  24. func (s *InMemorySessionManager) RelayToAll(ctx context.Context, msg wire.SNACMessage) {
  25. s.mapMutex.RLock()
  26. defer s.mapMutex.RUnlock()
  27. for _, sess := range s.store {
  28. s.maybeRelayMessage(ctx, msg, sess)
  29. }
  30. }
  31. // RelayToAllExcept relays a message to all session in the pool except for one
  32. // particular session.
  33. func (s *InMemorySessionManager) RelayToAllExcept(ctx context.Context, except *Session, msg wire.SNACMessage) {
  34. s.mapMutex.RLock()
  35. defer s.mapMutex.RUnlock()
  36. for _, sess := range s.store {
  37. if sess == except {
  38. continue
  39. }
  40. s.maybeRelayMessage(ctx, msg, sess)
  41. }
  42. }
  43. // RelayToScreenName relays a message to a session with a matching screen name.
  44. func (s *InMemorySessionManager) RelayToScreenName(ctx context.Context, screenName string, msg wire.SNACMessage) {
  45. sess := s.RetrieveByScreenName(screenName)
  46. if sess == nil {
  47. s.logger.WarnContext(ctx, "can't send notification because user is not online", "recipient", screenName, "message", msg)
  48. return
  49. }
  50. s.maybeRelayMessage(ctx, msg, sess)
  51. }
  52. // RelayToScreenNames relays a message to sessions with matching screenNames.
  53. func (s *InMemorySessionManager) RelayToScreenNames(ctx context.Context, screenNames []string, msg wire.SNACMessage) {
  54. for _, sess := range s.retrieveByScreenNames(screenNames) {
  55. s.maybeRelayMessage(ctx, msg, sess)
  56. }
  57. }
  58. func (s *InMemorySessionManager) maybeRelayMessage(ctx context.Context, msg wire.SNACMessage, sess *Session) {
  59. switch sess.RelayMessage(msg) {
  60. case SessSendClosed:
  61. s.logger.WarnContext(ctx, "can't send notification because the user's session is closed", "recipient", sess.ScreenName(), "message", msg)
  62. case SessQueueFull:
  63. s.logger.WarnContext(ctx, "can't send notification because queue is full", "recipient", sess.ScreenName(), "message", msg)
  64. sess.Close()
  65. }
  66. }
  67. // AddSession adds a new session to the pool. It replaces an existing session
  68. // with a matching screen name, ensuring that each screen name is unique in the
  69. // pool.
  70. func (s *InMemorySessionManager) AddSession(sessID string, screenName string) *Session {
  71. s.mapMutex.Lock()
  72. defer s.mapMutex.Unlock()
  73. // Only allow one session at a time per screen name. A session may already
  74. // exist because:
  75. // 1) the user is signing on using an already logged-on screen name.
  76. // 2) the session might be orphaned due to an undetected client
  77. // disconnection.
  78. for _, sess := range s.store {
  79. if screenName == sess.ScreenName() {
  80. sess.Close()
  81. delete(s.store, sess.ID())
  82. break
  83. }
  84. }
  85. sess := NewSession()
  86. sess.SetID(sessID)
  87. sess.SetScreenName(screenName)
  88. s.store[sess.ID()] = sess
  89. return sess
  90. }
  91. // RemoveSession takes a session out of the session pool.
  92. func (s *InMemorySessionManager) RemoveSession(sess *Session) {
  93. s.mapMutex.Lock()
  94. defer s.mapMutex.Unlock()
  95. delete(s.store, sess.ID())
  96. }
  97. // RetrieveSession finds a session with a matching sessionID. Returns nil if
  98. // session is not found.
  99. func (s *InMemorySessionManager) RetrieveSession(sessionID string) *Session {
  100. s.mapMutex.RLock()
  101. defer s.mapMutex.RUnlock()
  102. return s.store[sessionID]
  103. }
  104. // RetrieveByScreenName find a session with a matching screen name. Returns nil
  105. // if session is not found.
  106. func (s *InMemorySessionManager) RetrieveByScreenName(screenName string) *Session {
  107. s.mapMutex.RLock()
  108. defer s.mapMutex.RUnlock()
  109. for _, sess := range s.store {
  110. if screenName == sess.ScreenName() {
  111. return sess
  112. }
  113. }
  114. return nil
  115. }
  116. func (s *InMemorySessionManager) retrieveByScreenNames(screenNames []string) []*Session {
  117. s.mapMutex.RLock()
  118. defer s.mapMutex.RUnlock()
  119. var ret []*Session
  120. for _, sn := range screenNames {
  121. for _, sess := range s.store {
  122. if sn == sess.ScreenName() {
  123. ret = append(ret, sess)
  124. }
  125. }
  126. }
  127. return ret
  128. }
  129. // Empty returns true if the session pool contains 0 sessions.
  130. func (s *InMemorySessionManager) Empty() bool {
  131. s.mapMutex.RLock()
  132. defer s.mapMutex.RUnlock()
  133. return len(s.store) == 0
  134. }
  135. // AllSessions returns all sessions in the session pool.
  136. func (s *InMemorySessionManager) AllSessions() []*Session {
  137. s.mapMutex.RLock()
  138. defer s.mapMutex.RUnlock()
  139. var sessions []*Session
  140. for _, sess := range s.store {
  141. sessions = append(sessions, sess)
  142. }
  143. return sessions
  144. }