mirror of
https://github.com/truewhile/MeBox.git
synced 2026-10-05 21:06:38 +08:00
fix: refresh emby sessions in realtime
This commit is contained in:
@@ -6,6 +6,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
||||
)
|
||||
|
||||
@@ -61,8 +63,9 @@ func TestBotAdminUnbindInactiveAndInvalidBindings(t *testing.T) {
|
||||
recentTime := time.Now().Add(-2 * 24 * time.Hour)
|
||||
admin := &model.User{Username: "root", PasswordHash: "x", Role: "admin", IsActive: true, LastLoginAt: &oldTime}
|
||||
oldUser := &model.User{Username: "old", PasswordHash: "x", Role: "user", IsActive: true, LastLoginAt: &oldTime}
|
||||
realtimeUser := &model.User{Username: "realtime", PasswordHash: "x", Role: "user", IsActive: true, LastLoginAt: &oldTime}
|
||||
recentUser := &model.User{Username: "recent", PasswordHash: "x", Role: "user", IsActive: true, LastLoginAt: &recentTime}
|
||||
for _, user := range []*model.User{admin, oldUser, recentUser} {
|
||||
for _, user := range []*model.User{admin, oldUser, realtimeUser, recentUser} {
|
||||
if err := repos.User.Create(ctx, user); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -71,6 +74,7 @@ func TestBotAdminUnbindInactiveAndInvalidBindings(t *testing.T) {
|
||||
{TelegramUserID: 9501, TelegramName: "@root", ChatID: 9501, UserID: admin.ID},
|
||||
{TelegramUserID: 9502, TelegramName: "@old", ChatID: 9502, UserID: oldUser.ID},
|
||||
{TelegramUserID: 9503, TelegramName: "@recent", ChatID: 9503, UserID: recentUser.ID},
|
||||
{TelegramUserID: 9505, TelegramName: "@realtime", ChatID: 9505, UserID: realtimeUser.ID},
|
||||
{TelegramUserID: 9504, TelegramName: "@ghost", ChatID: 9504, UserID: "missing-user"},
|
||||
} {
|
||||
row := binding
|
||||
@@ -80,6 +84,11 @@ func TestBotAdminUnbindInactiveAndInvalidBindings(t *testing.T) {
|
||||
}
|
||||
channel := &model.NotifyChannel{Name: "Telegram", Type: "telegram", Enabled: true, Config: `{"admin_user_ids":"9501"}`}
|
||||
msg := &TelegramMessage{From: TelegramUser{ID: 9501, Username: "root"}, Chat: TelegramChat{ID: 9501, Type: "private"}}
|
||||
tracker := NewSessionTrackerService(zap.NewNop())
|
||||
tracker.RecordActivity(ctx, realtimeUser.ID, realtimeUser.Username, "phone-1", "iPhone", "Infuse", "192.0.2.10")
|
||||
device := NewDeviceService(zap.NewNop(), repos)
|
||||
device.SetSessionTracker(tracker)
|
||||
bot.SetDeviceService(device)
|
||||
|
||||
reply, err := bot.executeCommand(ctx, channel, msg, "/unbind_inactive 30")
|
||||
if err != nil {
|
||||
@@ -97,6 +106,9 @@ func TestBotAdminUnbindInactiveAndInvalidBindings(t *testing.T) {
|
||||
if binding := bot.telegramBinding(ctx, 9503); binding == nil {
|
||||
t.Fatal("recent user binding should remain")
|
||||
}
|
||||
if binding := bot.telegramBinding(ctx, 9505); binding == nil {
|
||||
t.Fatal("realtime active user binding should remain")
|
||||
}
|
||||
|
||||
reply, err = bot.executeCommand(ctx, channel, msg, "/unbind_duplicates")
|
||||
if err != nil {
|
||||
|
||||
@@ -70,6 +70,7 @@ type SessionTrackerService struct {
|
||||
|
||||
mu sync.RWMutex
|
||||
sessions map[string]RealtimeSession
|
||||
activity map[string]time.Time
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
@@ -77,6 +78,7 @@ func NewSessionTrackerService(log *zap.Logger) *SessionTrackerService {
|
||||
return &SessionTrackerService{
|
||||
log: log,
|
||||
sessions: make(map[string]RealtimeSession),
|
||||
activity: make(map[string]time.Time),
|
||||
now: time.Now,
|
||||
}
|
||||
}
|
||||
@@ -130,6 +132,10 @@ func (s *SessionTrackerService) Logout(ctx context.Context, userID, deviceID, re
|
||||
remoteEndPoint = strings.TrimSpace(remoteEndPoint)
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.activity == nil {
|
||||
s.activity = make(map[string]time.Time)
|
||||
}
|
||||
s.activity[userID] = s.now()
|
||||
for key, sess := range s.sessions {
|
||||
if sess.UserID != userID {
|
||||
continue
|
||||
@@ -208,6 +214,17 @@ func (s *SessionTrackerService) activityByUser(ctx context.Context) map[string]u
|
||||
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) == "" {
|
||||
@@ -231,6 +248,16 @@ func (s *SessionTrackerService) activityByUser(ctx context.Context) map[string]u
|
||||
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 == "" {
|
||||
@@ -248,6 +275,10 @@ func (s *SessionTrackerService) upsert(ctx context.Context, in realtimeSessionIn
|
||||
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
|
||||
@@ -315,6 +346,11 @@ func (s *SessionTrackerService) pruneLocked(now time.Time) {
|
||||
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 {
|
||||
|
||||
@@ -77,6 +77,26 @@ func TestActivityRefreshKeepsPlaybackState(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLogoutKeepsRealtimeLastActivityWithoutOnlineSession(t *testing.T) {
|
||||
tracker := NewSessionTrackerService(zap.NewNop())
|
||||
now := time.Date(2026, 6, 21, 12, 30, 0, 0, time.UTC)
|
||||
tracker.now = func() time.Time { return now }
|
||||
old := now.Add(-8 * time.Hour)
|
||||
users := []model.User{{Base: model.Base{ID: "u1"}, Username: "viewer", LastLoginAt: &old}}
|
||||
|
||||
tracker.RecordActivity(t.Context(), "u1", "viewer", "dev-1", "iPhone", "Infuse", "10.0.0.8")
|
||||
now = now.Add(time.Minute)
|
||||
tracker.Logout(t.Context(), "u1", "dev-1", "10.0.0.8")
|
||||
tracker.ApplyToUsers(t.Context(), users)
|
||||
|
||||
if users[0].LastLoginAt == nil || !users[0].LastLoginAt.Equal(now) {
|
||||
t.Fatalf("last_login_at = %v, want logout activity %v", users[0].LastLoginAt, now)
|
||||
}
|
||||
if users[0].RealtimeOnline || users[0].RealtimeDeviceCount != 0 {
|
||||
t.Fatalf("logged-out user should keep last activity but no online devices, online=%v devices=%d", users[0].RealtimeOnline, users[0].RealtimeDeviceCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBotDevicesIncludesRealtimeSessionOnlyDevices(t *testing.T) {
|
||||
repos, bot := newBotTestService(t)
|
||||
user := model.User{Base: model.Base{ID: "u1"}, Username: "viewer", PasswordHash: "x", Role: "user", IsActive: true}
|
||||
@@ -107,6 +127,31 @@ func TestBotDevicesIncludesRealtimeSessionOnlyDevices(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBotUserInfoUsesRealtimeLastLogin(t *testing.T) {
|
||||
repos, bot := newBotTestService(t)
|
||||
now := time.Date(2026, 6, 21, 13, 45, 0, 0, time.UTC)
|
||||
old := now.Add(-6 * time.Hour)
|
||||
user := model.User{Base: model.Base{ID: "u1"}, Username: "viewer", PasswordHash: "x", Role: "user", IsActive: true, LastLoginAt: &old}
|
||||
if err := repos.User.Create(t.Context(), &user); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tracker := NewSessionTrackerService(zap.NewNop())
|
||||
tracker.now = func() time.Time { return now }
|
||||
tracker.RecordActivity(t.Context(), user.ID, user.Username, "dev-1", "Apple TV", "Yamby", "10.0.0.8")
|
||||
device := NewDeviceService(zap.NewNop(), repos)
|
||||
device.SetSessionTracker(tracker)
|
||||
bot.SetDeviceService(device)
|
||||
|
||||
reply := bot.cmdMgoUserInfo(t.Context(), []string{"viewer"})
|
||||
|
||||
if !strings.Contains(reply.Text, "最后登录:<b>2026-06-21 13:45</b>") {
|
||||
t.Fatalf("reply should use realtime last login, got %q", reply.Text)
|
||||
}
|
||||
if !strings.Contains(reply.Text, "设备:<b>1</b>") {
|
||||
t.Fatalf("reply should count realtime device, got %q", reply.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRealtimeRecentLoginProtectsCleanupCandidate(t *testing.T) {
|
||||
repos := newSessionTrackerTestRepos(t)
|
||||
now := time.Date(2026, 6, 21, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
@@ -43,6 +43,7 @@ func (s *TelegramBotService) cmdMgoUserInfo(ctx context.Context, args []string)
|
||||
if user == nil {
|
||||
return telegramCommandReply{Text: "未找到用户。"}
|
||||
}
|
||||
s.applyRealtimeUserActivity(ctx, user)
|
||||
devices, _ := s.listUserDevices(ctx, user.ID)
|
||||
var historyCount int64
|
||||
_ = s.repo.DB.WithContext(ctx).Model(&model.PlaybackHistory{}).Where("user_id = ?", user.ID).Count(&historyCount).Error
|
||||
@@ -60,6 +61,15 @@ func (s *TelegramBotService) cmdMgoUserInfo(ctx context.Context, args []string)
|
||||
)}
|
||||
}
|
||||
|
||||
func (s *TelegramBotService) applyRealtimeUserActivity(ctx context.Context, user *model.User) {
|
||||
if s == nil || user == nil || s.device == nil || s.device.sessions == nil {
|
||||
return
|
||||
}
|
||||
users := []model.User{*user}
|
||||
s.device.sessions.ApplyToUsers(ctx, users)
|
||||
*user = users[0]
|
||||
}
|
||||
|
||||
func (s *TelegramBotService) cmdMgoDeleteUser(ctx context.Context, args []string) telegramCommandReply {
|
||||
if len(args) < 2 || !strings.EqualFold(args[len(args)-1], "confirm") {
|
||||
return telegramCommandReply{Text: "删除用户需要确认:<code>/rmemby 用户名 confirm</code> 或 <code>/urm 用户名 confirm</code>"}
|
||||
|
||||
@@ -119,6 +119,7 @@ func (s *TelegramBotService) cmdUnbindInactive(ctx context.Context, args []strin
|
||||
return telegramCommandReply{Text: "读取用户失败:" + err.Error()}
|
||||
}
|
||||
cutoff := time.Now().Add(-time.Duration(days) * 24 * time.Hour)
|
||||
recentWindow := time.Duration(days) * 24 * time.Hour
|
||||
var userIDs []string
|
||||
var done []string
|
||||
for _, user := range users {
|
||||
@@ -132,6 +133,9 @@ func (s *TelegramBotService) cmdUnbindInactive(ctx context.Context, args []strin
|
||||
if lastActive.IsZero() || lastActive.After(cutoff) {
|
||||
continue
|
||||
}
|
||||
if s.device != nil && s.device.UserRecentlyActive(ctx, user.ID, recentWindow) {
|
||||
continue
|
||||
}
|
||||
var count int64
|
||||
_ = s.repo.DB.WithContext(ctx).Model(&model.TelegramBinding{}).Where("user_id = ?", user.ID).Count(&count).Error
|
||||
if count == 0 {
|
||||
|
||||
Reference in New Issue
Block a user