diff --git a/internal/service/session_tracker.go b/internal/service/session_tracker.go index 6fcce34..f4575cc 100644 --- a/internal/service/session_tracker.go +++ b/internal/service/session_tracker.go @@ -8,8 +8,6 @@ import ( "time" "go.uber.org/zap" - - "github.com/ShukeBta/MediaStationGo/internal/model" ) const ( @@ -182,207 +180,3 @@ func (s *SessionTrackerService) ListByUser(ctx context.Context, userID string) [ } return out } - -func (s *SessionTrackerService) ApplyToUsers(ctx context.Context, users []model.User) { - if s == nil || len(users) == 0 { - return - } - activity := s.activityByUser(ctx) - for i := range users { - a, ok := activity[users[i].ID] - if !ok { - continue - } - if a.LastActivityAt != nil && (users[i].LastLoginAt == nil || a.LastActivityAt.After(*users[i].LastLoginAt)) { - t := *a.LastActivityAt - users[i].LastLoginAt = &t - } - users[i].RealtimeOnline = a.Online - users[i].RealtimeDeviceCount = a.ActiveDeviceCount - } -} - -func (s *SessionTrackerService) UserRecentlyActive(ctx context.Context, userID string, within time.Duration) bool { - if s == nil || within <= 0 { - return false - } - activity := s.activityByUser(ctx)[strings.TrimSpace(userID)] - return activity.LastActivityAt != nil && activity.LastActivityAt.After(s.now().Add(-within)) -} - -func (s *SessionTrackerService) activityByUser(ctx context.Context) map[string]userRealtimeActivity { - sessions := s.List(ctx) - now := s.now() - out := make(map[string]userRealtimeActivity) - for userID, lastActivity := range s.userActivitySnapshot() { - if strings.TrimSpace(userID) == "" { - continue - } - a := out[userID] - if a.LastActivityAt == nil || lastActivity.After(*a.LastActivityAt) { - t := lastActivity - a.LastActivityAt = &t - } - out[userID] = a - } - seenDevices := make(map[string]map[string]struct{}) - for _, sess := range sessions { - if strings.TrimSpace(sess.UserID) == "" { - continue - } - a := out[sess.UserID] - if a.LastActivityAt == nil || sess.LastActivityAt.After(*a.LastActivityAt) { - t := sess.LastActivityAt - a.LastActivityAt = &t - } - if sess.LastActivityAt.After(now.Add(-realtimeSessionOnlineTTL)) { - a.Online = true - } - if seenDevices[sess.UserID] == nil { - seenDevices[sess.UserID] = map[string]struct{}{} - } - seenDevices[sess.UserID][sessionDeviceKey(sess)] = struct{}{} - a.ActiveDeviceCount = len(seenDevices[sess.UserID]) - out[sess.UserID] = a - } - return out -} - -func (s *SessionTrackerService) userActivitySnapshot() map[string]time.Time { - s.mu.RLock() - defer s.mu.RUnlock() - out := make(map[string]time.Time, len(s.activity)) - for userID, lastActivity := range s.activity { - out[userID] = lastActivity - } - return out -} - -func (s *SessionTrackerService) upsert(ctx context.Context, in realtimeSessionInput) { - userID := strings.TrimSpace(in.UserID) - if userID == "" { - return - } - now := s.now() - in.DeviceID = strings.TrimSpace(in.DeviceID) - in.DeviceName = strings.TrimSpace(in.DeviceName) - in.Client = strings.TrimSpace(in.Client) - in.RemoteEndPoint = strings.TrimSpace(in.RemoteEndPoint) - if in.DeviceID == "" { - in.DeviceID = fallbackSessionDeviceID(in.DeviceName, in.Client, in.RemoteEndPoint) - } - key := userID + "\x00" + realtimeSessionTerminalKey(in.DeviceID, in.DeviceName, in.Client, in.RemoteEndPoint) - s.mu.Lock() - defer s.mu.Unlock() - s.pruneLocked(now) - if s.activity == nil { - s.activity = make(map[string]time.Time) - } - s.activity[userID] = now - existing, existed := s.sessions[key] - if strings.TrimSpace(in.UserName) == "" { - in.UserName = existing.UserName - } - if in.DeviceName == "" { - in.DeviceName = existing.DeviceName - } - if in.Client == "" { - in.Client = existing.Client - } - if in.RemoteEndPoint == "" { - in.RemoteEndPoint = existing.RemoteEndPoint - } - lastPlaybackAt := existing.LastPlaybackAt - itemID := existing.ItemID - positionTicks := existing.PositionTicks - runtimeTicks := existing.RuntimeTicks - isPlaying := existing.IsPlaying - isPaused := existing.IsPaused - if in.PlaybackUpdate { - itemID = firstNonEmptyString(in.ItemID, existing.ItemID) - if in.ItemID != "" || in.PositionTicks != 0 { - positionTicks = in.PositionTicks - } - if in.ItemID != "" || in.RuntimeTicks != 0 { - runtimeTicks = in.RuntimeTicks - } - isPlaying = in.IsPlaying - isPaused = in.IsPaused - } - if in.PlaybackUpdate && (in.ItemID != "" || in.IsPlaying) { - t := now - lastPlaybackAt = &t - } - s.sessions[key] = RealtimeSession{ - ID: key, - UserID: userID, - UserName: strings.TrimSpace(in.UserName), - DeviceID: in.DeviceID, - DeviceName: in.DeviceName, - Client: in.Client, - RemoteEndPoint: in.RemoteEndPoint, - LastActivityAt: now, - ItemID: itemID, - PositionTicks: positionTicks, - RuntimeTicks: runtimeTicks, - IsPlaying: isPlaying, - IsPaused: isPaused, - LastPlaybackAt: lastPlaybackAt, - } - if !existed && s.log != nil { - s.log.Debug("realtime session started", - zap.String("user_id", userID), - zap.String("device_id", in.DeviceID), - zap.String("client", in.Client), - zap.String("remote", in.RemoteEndPoint), - ) - } -} - -func (s *SessionTrackerService) pruneLocked(now time.Time) { - expiresBefore := now.Add(-realtimeSessionTTL) - for key, sess := range s.sessions { - if sess.LastActivityAt.Before(expiresBefore) { - delete(s.sessions, key) - } - } - for userID, lastActivity := range s.activity { - if lastActivity.Before(expiresBefore) { - delete(s.activity, userID) - } - } -} - -func fallbackSessionDeviceID(deviceName, client, remoteEndPoint string) string { - parts := []string{strings.TrimSpace(deviceName), strings.TrimSpace(client), strings.TrimSpace(remoteEndPoint)} - joined := strings.Trim(strings.Join(parts, "|"), "|") - if joined == "" { - joined = "unknown" - } - return "rt-" + fingerprint(client, joined) -} - -func sessionDeviceKey(sess RealtimeSession) string { - return realtimeSessionTerminalKey(sess.DeviceID, sess.DeviceName, sess.Client, sess.RemoteEndPoint) -} - -func realtimeSessionTerminalKey(deviceID, deviceName, client, remoteEndPoint string) string { - deviceName = strings.TrimSpace(deviceName) - if deviceName != "" { - return "fp-" + fingerprint(client, deviceName) - } - deviceID = strings.TrimSpace(deviceID) - if deviceID != "" { - return deviceID - } - return fallbackSessionDeviceID(deviceName, client, remoteEndPoint) -} - -func firstNonEmptyString(values ...string) string { - for _, value := range values { - if strings.TrimSpace(value) != "" { - return strings.TrimSpace(value) - } - } - return "" -} diff --git a/internal/service/session_tracker_activity.go b/internal/service/session_tracker_activity.go new file mode 100644 index 0000000..00a97d4 --- /dev/null +++ b/internal/service/session_tracker_activity.go @@ -0,0 +1,84 @@ +package service + +import ( + "context" + "strings" + "time" + + "github.com/ShukeBta/MediaStationGo/internal/model" +) + +func (s *SessionTrackerService) ApplyToUsers(ctx context.Context, users []model.User) { + if s == nil || len(users) == 0 { + return + } + activity := s.activityByUser(ctx) + for i := range users { + a, ok := activity[users[i].ID] + if !ok { + continue + } + if a.LastActivityAt != nil && (users[i].LastLoginAt == nil || a.LastActivityAt.After(*users[i].LastLoginAt)) { + t := *a.LastActivityAt + users[i].LastLoginAt = &t + } + users[i].RealtimeOnline = a.Online + users[i].RealtimeDeviceCount = a.ActiveDeviceCount + } +} + +func (s *SessionTrackerService) UserRecentlyActive(ctx context.Context, userID string, within time.Duration) bool { + if s == nil || within <= 0 { + return false + } + activity := s.activityByUser(ctx)[strings.TrimSpace(userID)] + return activity.LastActivityAt != nil && activity.LastActivityAt.After(s.now().Add(-within)) +} + +func (s *SessionTrackerService) activityByUser(ctx context.Context) map[string]userRealtimeActivity { + sessions := s.List(ctx) + now := s.now() + out := make(map[string]userRealtimeActivity) + for userID, lastActivity := range s.userActivitySnapshot() { + if strings.TrimSpace(userID) == "" { + continue + } + a := out[userID] + if a.LastActivityAt == nil || lastActivity.After(*a.LastActivityAt) { + t := lastActivity + a.LastActivityAt = &t + } + out[userID] = a + } + seenDevices := make(map[string]map[string]struct{}) + for _, sess := range sessions { + if strings.TrimSpace(sess.UserID) == "" { + continue + } + a := out[sess.UserID] + if a.LastActivityAt == nil || sess.LastActivityAt.After(*a.LastActivityAt) { + t := sess.LastActivityAt + a.LastActivityAt = &t + } + if sess.LastActivityAt.After(now.Add(-realtimeSessionOnlineTTL)) { + a.Online = true + } + if seenDevices[sess.UserID] == nil { + seenDevices[sess.UserID] = map[string]struct{}{} + } + seenDevices[sess.UserID][sessionDeviceKey(sess)] = struct{}{} + a.ActiveDeviceCount = len(seenDevices[sess.UserID]) + out[sess.UserID] = a + } + return out +} + +func (s *SessionTrackerService) userActivitySnapshot() map[string]time.Time { + s.mu.RLock() + defer s.mu.RUnlock() + out := make(map[string]time.Time, len(s.activity)) + for userID, lastActivity := range s.activity { + out[userID] = lastActivity + } + return out +} diff --git a/internal/service/session_tracker_keys.go b/internal/service/session_tracker_keys.go new file mode 100644 index 0000000..f236238 --- /dev/null +++ b/internal/service/session_tracker_keys.go @@ -0,0 +1,37 @@ +package service + +import "strings" + +func fallbackSessionDeviceID(deviceName, client, remoteEndPoint string) string { + parts := []string{strings.TrimSpace(deviceName), strings.TrimSpace(client), strings.TrimSpace(remoteEndPoint)} + joined := strings.Trim(strings.Join(parts, "|"), "|") + if joined == "" { + joined = "unknown" + } + return "rt-" + fingerprint(client, joined) +} + +func sessionDeviceKey(sess RealtimeSession) string { + return realtimeSessionTerminalKey(sess.DeviceID, sess.DeviceName, sess.Client, sess.RemoteEndPoint) +} + +func realtimeSessionTerminalKey(deviceID, deviceName, client, remoteEndPoint string) string { + deviceName = strings.TrimSpace(deviceName) + if deviceName != "" { + return "fp-" + fingerprint(client, deviceName) + } + deviceID = strings.TrimSpace(deviceID) + if deviceID != "" { + return deviceID + } + return fallbackSessionDeviceID(deviceName, client, remoteEndPoint) +} + +func firstNonEmptyString(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return strings.TrimSpace(value) + } + } + return "" +} diff --git a/internal/service/session_tracker_store.go b/internal/service/session_tracker_store.go new file mode 100644 index 0000000..84654c7 --- /dev/null +++ b/internal/service/session_tracker_store.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "strings" + "time" + + "go.uber.org/zap" +) + +func (s *SessionTrackerService) upsert(ctx context.Context, in realtimeSessionInput) { + userID := strings.TrimSpace(in.UserID) + if userID == "" { + return + } + now := s.now() + in.DeviceID = strings.TrimSpace(in.DeviceID) + in.DeviceName = strings.TrimSpace(in.DeviceName) + in.Client = strings.TrimSpace(in.Client) + in.RemoteEndPoint = strings.TrimSpace(in.RemoteEndPoint) + if in.DeviceID == "" { + in.DeviceID = fallbackSessionDeviceID(in.DeviceName, in.Client, in.RemoteEndPoint) + } + key := userID + "\x00" + realtimeSessionTerminalKey(in.DeviceID, in.DeviceName, in.Client, in.RemoteEndPoint) + s.mu.Lock() + defer s.mu.Unlock() + s.pruneLocked(now) + if s.activity == nil { + s.activity = make(map[string]time.Time) + } + s.activity[userID] = now + existing, existed := s.sessions[key] + if strings.TrimSpace(in.UserName) == "" { + in.UserName = existing.UserName + } + if in.DeviceName == "" { + in.DeviceName = existing.DeviceName + } + if in.Client == "" { + in.Client = existing.Client + } + if in.RemoteEndPoint == "" { + in.RemoteEndPoint = existing.RemoteEndPoint + } + lastPlaybackAt := existing.LastPlaybackAt + itemID := existing.ItemID + positionTicks := existing.PositionTicks + runtimeTicks := existing.RuntimeTicks + isPlaying := existing.IsPlaying + isPaused := existing.IsPaused + if in.PlaybackUpdate { + itemID = firstNonEmptyString(in.ItemID, existing.ItemID) + if in.ItemID != "" || in.PositionTicks != 0 { + positionTicks = in.PositionTicks + } + if in.ItemID != "" || in.RuntimeTicks != 0 { + runtimeTicks = in.RuntimeTicks + } + isPlaying = in.IsPlaying + isPaused = in.IsPaused + } + if in.PlaybackUpdate && (in.ItemID != "" || in.IsPlaying) { + t := now + lastPlaybackAt = &t + } + s.sessions[key] = RealtimeSession{ + ID: key, + UserID: userID, + UserName: strings.TrimSpace(in.UserName), + DeviceID: in.DeviceID, + DeviceName: in.DeviceName, + Client: in.Client, + RemoteEndPoint: in.RemoteEndPoint, + LastActivityAt: now, + ItemID: itemID, + PositionTicks: positionTicks, + RuntimeTicks: runtimeTicks, + IsPlaying: isPlaying, + IsPaused: isPaused, + LastPlaybackAt: lastPlaybackAt, + } + if !existed && s.log != nil { + s.log.Debug("realtime session started", + zap.String("user_id", userID), + zap.String("device_id", in.DeviceID), + zap.String("client", in.Client), + zap.String("remote", in.RemoteEndPoint), + ) + } +} + +func (s *SessionTrackerService) pruneLocked(now time.Time) { + expiresBefore := now.Add(-realtimeSessionTTL) + for key, sess := range s.sessions { + if sess.LastActivityAt.Before(expiresBefore) { + delete(s.sessions, key) + } + } + for userID, lastActivity := range s.activity { + if lastActivity.Before(expiresBefore) { + delete(s.activity, userID) + } + } +}