mirror of
https://github.com/truewhile/MeBox.git
synced 2026-10-06 05:16:38 +08:00
split session tracker helpers
This commit is contained in:
@@ -8,8 +8,6 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
|
|
||||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@@ -182,207 +180,3 @@ func (s *SessionTrackerService) ListByUser(ctx context.Context, userID string) [
|
|||||||
}
|
}
|
||||||
return out
|
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 ""
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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 ""
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user