root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(desktop): report current user online clients

Browse Source
This commit is contained in:
liangxu
2026-05-06 18:25:05 +08:00
parent 930f095c64
commit 47681ab1f5
12 changed files with 297 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/desktop_client_service.go
+56 -9
View File
@@ -84,8 +84,10 @@ type RotateDesktopClientResponse struct {
}
type HeartbeatDesktopClientResponse struct {
Client DesktopClientView `json:"client"`
ServerTime time.Time `json:"server_time"`
Client DesktopClientView `json:"client"`
ServerTime time.Time `json:"server_time"`
OnlineClientCount int `json:"online_client_count"`
CurrentUserOnlineClientCount int `json:"current_user_online_client_count"`
}
type OfflineDesktopClientResponse struct {
@@ -128,7 +130,7 @@ func (s *DesktopClientService) Register(ctx context.Context, actor auth.Actor, r
return nil, response.ErrInternal(50072, "desktop_client_register_failed", "failed to persist desktop client")
}
markDesktopClientPresent(ctx, s.redis, client.ID)
markDesktopClientPresentForUser(ctx, s.redis, client.ID, client.TenantID, client.WorkspaceID, client.UserID)
return &RegisterDesktopClientResponse{
ClientID: client.ID.String(),
@@ -160,7 +162,7 @@ func (s *DesktopClientService) Rotate(ctx context.Context, client *repository.De
return nil, response.ErrInternal(50074, "desktop_client_rotate_failed", "failed to rotate desktop client token")
}
markDesktopClientPresent(ctx, s.redis, updated.ID)
markDesktopClientPresentForUser(ctx, s.redis, updated.ID, updated.TenantID, updated.WorkspaceID, updated.UserID)
return &RotateDesktopClientResponse{
ClientToken: rawToken,
@@ -190,8 +192,12 @@ func (s *DesktopClientService) Heartbeat(ctx context.Context, client *repository
return nil, response.ErrInternal(50075, "desktop_client_heartbeat_failed", "failed to persist desktop client heartbeat")
}
markDesktopClientPresent(ctx, s.redis, updated.ID)
markDesktopClientPresentForUser(ctx, s.redis, updated.ID, updated.TenantID, updated.WorkspaceID, updated.UserID)
markDesktopAccountsPresent(ctx, s.redis, updated.ID, parseDesktopAccountIDs(req.AccountIDs))
onlineClientCount, err := s.currentUserOnlineClientCount(ctx, updated)
if err != nil {
return nil, err
}
if req.Startup && s.taskSvc != nil {
if err := s.taskSvc.recoverClientTasksOnStartup(ctx, updated); err != nil {
@@ -200,8 +206,10 @@ func (s *DesktopClientService) Heartbeat(ctx context.Context, client *repository
}
return &HeartbeatDesktopClientResponse{
Client: buildDesktopClientView(updated),
ServerTime: time.Now().UTC(),
Client: buildDesktopClientView(updated),
ServerTime: time.Now().UTC(),
OnlineClientCount: onlineClientCount,
CurrentUserOnlineClientCount: onlineClientCount,
}, nil
}
@@ -218,7 +226,7 @@ func (s *DesktopClientService) Revoke(ctx context.Context, client *repository.De
return nil, response.ErrInternal(50076, "desktop_client_revoke_failed", "failed to revoke desktop client")
}
clearDesktopClientPresenceState(ctx, s.redis, updated.ID)
clearDesktopClientPresenceStateForUser(ctx, s.redis, updated.ID, updated.TenantID, updated.WorkspaceID, updated.UserID)
if s.taskSvc != nil {
if err := s.taskSvc.recoverClientTasksOnDisconnect(ctx, updated); err != nil {
return nil, err
@@ -234,7 +242,7 @@ func (s *DesktopClientService) Offline(ctx context.Context, client *repository.D
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
clearDesktopClientPresenceState(ctx, s.redis, client.ID)
clearDesktopClientPresenceStateForUser(ctx, s.redis, client.ID, client.TenantID, client.WorkspaceID, client.UserID)
if s.taskSvc != nil {
if err := s.taskSvc.recoverClientTasksOnDisconnect(ctx, client); err != nil {
return nil, err
@@ -246,6 +254,45 @@ func (s *DesktopClientService) Offline(ctx context.Context, client *repository.D
}, nil
}
func (s *DesktopClientService) currentUserOnlineClientCount(ctx context.Context, client *repository.DesktopClient) (int, error) {
if count, ok := loadDesktopUserOnlineClientCount(ctx, s.redis, client.TenantID, client.WorkspaceID, client.UserID, time.Now().UTC()); ok {
return count, nil
}
clients, err := s.repo.ListByUser(ctx, client.TenantID, client.WorkspaceID, client.UserID)
if err != nil {
return 0, response.ErrInternal(50077, "desktop_client_online_count_failed", "failed to inspect online desktop clients")
}
return countOnlineDesktopClients(ctx, s.redis, clients, time.Now().UTC()), nil
}
func countOnlineDesktopClients(ctx context.Context, redis *goredis.Client, clients []repository.DesktopClient, now time.Time) int {
if len(clients) == 0 {
return 0
}
clientIDs := make([]uuid.UUID, 0, len(clients))
for _, client := range clients {
if client.ID == uuid.Nil || client.RevokedAt != nil {
continue
}
clientIDs = append(clientIDs, client.ID)
}
presence := loadDesktopClientPresence(ctx, redis, clientIDs)
onlineCount := 0
for _, client := range clients {
if client.ID == uuid.Nil || client.RevokedAt != nil {
continue
}
if resolveDesktopClientOnline(&client, presence[client.ID], presence != nil, now) {
onlineCount++
}
}
return onlineCount
}
func newDesktopClientToken() (string, []byte, error) {
raw := make([]byte, 32)
if _, err := rand.Read(raw); err != nil {
server/internal/tenant/app/desktop_presence.go
+56
View File
@@ -27,6 +27,13 @@ func desktopClientAccountsPresenceKey(clientID uuid.UUID) string {
return "desktop:presence:client_accounts:" + clientID.String()
}
func desktopUserClientsPresenceKey(tenantID, workspaceID, userID int64) string {
return "desktop:presence:user_clients:" +
strconv.FormatInt(tenantID, 10) + ":" +
strconv.FormatInt(workspaceID, 10) + ":" +
strconv.FormatInt(userID, 10)
}
func markDesktopClientPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) {
if redis == nil {
return
@@ -35,6 +42,25 @@ func markDesktopClientPresent(ctx context.Context, redis *goredis.Client, client
_ = redis.Set(ctx, desktopClientPresenceKey(clientID), time.Now().UTC().Format(time.RFC3339), desktopClientPresenceTTL).Err()
}
func markDesktopClientPresentForUser(ctx context.Context, redis *goredis.Client, clientID uuid.UUID, tenantID, workspaceID, userID int64) {
markDesktopClientPresent(ctx, redis, clientID)
if redis == nil || clientID == uuid.Nil || tenantID == 0 || workspaceID == 0 || userID == 0 {
return
}
now := time.Now().UTC()
staleBefore := strconv.FormatInt(now.Add(-desktopClientPresenceTTL).UnixMilli(), 10)
key := desktopUserClientsPresenceKey(tenantID, workspaceID, userID)
pipe := redis.Pipeline()
pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
pipe.ZAdd(ctx, key, goredis.Z{
Score: float64(now.UnixMilli()),
Member: clientID.String(),
})
pipe.Expire(ctx, key, desktopClientPresenceTTL*4)
_, _ = pipe.Exec(ctx)
}
func clearDesktopClientPresence(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) {
if redis == nil {
return
@@ -43,6 +69,20 @@ func clearDesktopClientPresence(ctx context.Context, redis *goredis.Client, clie
_ = redis.Del(ctx, desktopClientPresenceKey(clientID)).Err()
}
func clearDesktopClientPresenceStateForUser(ctx context.Context, redis *goredis.Client, clientID uuid.UUID, tenantID, workspaceID, userID int64) {
clearDesktopClientPresenceState(ctx, redis, clientID)
if redis == nil || clientID == uuid.Nil || tenantID == 0 || workspaceID == 0 || userID == 0 {
return
}
staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopClientPresenceTTL).UnixMilli(), 10)
key := desktopUserClientsPresenceKey(tenantID, workspaceID, userID)
pipe := redis.Pipeline()
pipe.ZRem(ctx, key, clientID.String())
pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
_, _ = pipe.Exec(ctx)
}
func loadDesktopClientPresence(ctx context.Context, redis *goredis.Client, clientIDs []uuid.UUID) map[uuid.UUID]bool {
if redis == nil || len(clientIDs) == 0 {
return nil
@@ -75,6 +115,22 @@ func loadDesktopClientPresence(ctx context.Context, redis *goredis.Client, clien
return result
}
func loadDesktopUserOnlineClientCount(ctx context.Context, redis *goredis.Client, tenantID, workspaceID, userID int64, now time.Time) (int, bool) {
if redis == nil || tenantID == 0 || workspaceID == 0 || userID == 0 {
return 0, false
}
staleBefore := strconv.FormatInt(now.UTC().Add(-desktopClientPresenceTTL).UnixMilli(), 10)
key := desktopUserClientsPresenceKey(tenantID, workspaceID, userID)
pipe := redis.Pipeline()
pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
countCmd := pipe.ZCard(ctx, key)
if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil {
return 0, false
}
return int(countCmd.Val()), true
}
func markDesktopAccountsPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID, accountIDs []uuid.UUID) {
if redis == nil {
return
server/internal/tenant/app/desktop_presence_test.go
+86
View File
@@ -0,0 +1,86 @@
package app
import (
"context"
"testing"
"time"
"github.com/alicebob/miniredis/v2"
"github.com/google/uuid"
goredis "github.com/redis/go-redis/v9"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
)
func TestDesktopUserOnlineClientCountPrefersRedisPresenceIndex(t *testing.T) {
ctx := context.Background()
server := miniredis.RunT(t)
redis := goredis.NewClient(&goredis.Options{Addr: server.Addr()})
t.Cleanup(func() {
require.NoError(t, redis.Close())
})
tenantID := int64(7)
workspaceID := int64(11)
userID := int64(23)
firstClientID := uuid.New()
secondClientID := uuid.New()
markDesktopClientPresentForUser(ctx, redis, firstClientID, tenantID, workspaceID, userID)
markDesktopClientPresentForUser(ctx, redis, secondClientID, tenantID, workspaceID, userID)
count, ok := loadDesktopUserOnlineClientCount(ctx, redis, tenantID, workspaceID, userID, time.Now().UTC())
require.True(t, ok)
assert.Equal(t, 2, count)
}
func TestDesktopClientServiceOnlineCountReturnsRedisCountBeforeRepoFallback(t *testing.T) {
ctx := context.Background()
server := miniredis.RunT(t)
redis := goredis.NewClient(&goredis.Options{Addr: server.Addr()})
t.Cleanup(func() {
require.NoError(t, redis.Close())
})
client := &repository.DesktopClient{
ID: uuid.New(),
TenantID: 7,
WorkspaceID: 11,
UserID: 23,
}
markDesktopClientPresentForUser(ctx, redis, client.ID, client.TenantID, client.WorkspaceID, client.UserID)
markDesktopClientPresentForUser(ctx, redis, uuid.New(), client.TenantID, client.WorkspaceID, client.UserID)
count, err := (&DesktopClientService{redis: redis}).currentUserOnlineClientCount(ctx, client)
require.NoError(t, err)
assert.Equal(t, 2, count)
}
func TestDesktopUserOnlineClientCountPrunesStaleRedisPresence(t *testing.T) {
ctx := context.Background()
server := miniredis.RunT(t)
redis := goredis.NewClient(&goredis.Options{Addr: server.Addr()})
t.Cleanup(func() {
require.NoError(t, redis.Close())
})
tenantID := int64(7)
workspaceID := int64(11)
userID := int64(23)
key := desktopUserClientsPresenceKey(tenantID, workspaceID, userID)
now := time.Date(2026, 5, 6, 12, 0, 0, 0, time.UTC)
require.NoError(t, redis.ZAdd(ctx, key,
goredis.Z{Score: float64(now.Add(-2 * desktopClientPresenceTTL).UnixMilli()), Member: uuid.NewString()},
goredis.Z{Score: float64(now.Add(-time.Second).UnixMilli()), Member: uuid.NewString()},
).Err())
count, ok := loadDesktopUserOnlineClientCount(ctx, redis, tenantID, workspaceID, userID, now)
require.True(t, ok)
assert.Equal(t, 1, count)
}