feat: enhance desktop client presence management and add monitoring for active platforms

server/internal/tenant/app/desktop_presence.go

@@ -16,6 +16,11 @@ const desktopClientPresenceTTL = 30 * time.Second
 const desktopAccountPresenceTTL = 30 * time.Second
 const desktopTaskConsumerPresenceTTL = 75 * time.Second
 
+type desktopAccountPresenceClient struct {
+	ClientID             uuid.UUID
+	OnlineSinceUnixMilli int64
+}
+
 func desktopClientPresenceKey(clientID uuid.UUID) string {
 	return "desktop:presence:client:" + clientID.String()
 }
@@ -44,7 +49,13 @@ func markDesktopClientPresent(ctx context.Context, redis *goredis.Client, client
 		return
 	}
 
-	_ = redis.Set(ctx, desktopClientPresenceKey(clientID), time.Now().UTC().Format(time.RFC3339), desktopClientPresenceTTL).Err()
+	key := desktopClientPresenceKey(clientID)
+	now := strconv.FormatInt(time.Now().UTC().UnixMilli(), 10)
+	set, err := redis.SetNX(ctx, key, now, desktopClientPresenceTTL).Result()
+	if err != nil || set {
+		return
+	}
+	_ = redis.Expire(ctx, key, desktopClientPresenceTTL).Err()
 }
 
 func markDesktopTaskConsumerPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) {
@@ -198,8 +209,11 @@ func markDesktopAccountsPresent(ctx context.Context, redis *goredis.Client, clie
 		nextAccountSet[accountID] = struct{}{}
 	}
 
-	now := float64(time.Now().UTC().UnixMilli())
-	staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopAccountPresenceTTL).UnixMilli(), 10)
+	now := time.Now().UTC()
+	onlineSinceUnixMilli := now.UnixMilli()
+	if value, err := redis.Get(ctx, desktopClientPresenceKey(clientID)).Result(); err == nil {
+		onlineSinceUnixMilli = parseDesktopPresenceUnixMilli(value, now)
+	}
 	clientAccountsKey := desktopClientAccountsPresenceKey(clientID)
 	clientMember := clientID.String()
 
@@ -222,9 +236,8 @@ func markDesktopAccountsPresent(ctx context.Context, redis *goredis.Client, clie
 	}
 	for _, accountID := range uniqueIDs {
 		key := desktopAccountPresenceKey(accountID)
-		pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
 		pipe.ZAdd(ctx, key, goredis.Z{
-			Score:  now,
+			Score:  float64(onlineSinceUnixMilli),
 			Member: clientMember,
 		})
 		pipe.Expire(ctx, key, desktopAccountPresenceTTL*4)
@@ -259,44 +272,145 @@ func clearDesktopClientPresenceState(ctx context.Context, redis *goredis.Client,
 }
 
 func loadDesktopAccountPresence(ctx context.Context, redis *goredis.Client, accountIDs []uuid.UUID) map[uuid.UUID]uuid.UUID {
+	clientsByAccount := loadDesktopAccountPresenceClients(ctx, redis, accountIDs)
+	if len(clientsByAccount) == 0 {
+		return nil
+	}
+
+	result := make(map[uuid.UUID]uuid.UUID, len(clientsByAccount))
+	for accountID, clients := range clientsByAccount {
+		if len(clients) == 0 {
+			continue
+		}
+		result[accountID] = clients[len(clients)-1].ClientID
+	}
+	return result
+}
+
+func loadDesktopAccountPresenceClients(ctx context.Context, redis *goredis.Client, accountIDs []uuid.UUID) map[uuid.UUID][]desktopAccountPresenceClient {
 	if redis == nil || len(accountIDs) == 0 {
 		return nil
 	}
 
 	uniqueIDs := uniqueUUIDs(accountIDs)
-	staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopAccountPresenceTTL).UnixMilli(), 10)
 
 	pipe := redis.Pipeline()
 	commands := make(map[uuid.UUID]*goredis.ZSliceCmd, len(uniqueIDs))
 	for _, accountID := range uniqueIDs {
 		key := desktopAccountPresenceKey(accountID)
-		pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
-		commands[accountID] = pipe.ZRevRangeWithScores(ctx, key, 0, 0)
+		commands[accountID] = pipe.ZRangeWithScores(ctx, key, 0, -1)
 	}
 
 	if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil {
 		return nil
 	}
 
-	result := make(map[uuid.UUID]uuid.UUID, len(commands))
+	result := make(map[uuid.UUID][]desktopAccountPresenceClient, len(commands))
+	clientIDs := make([]uuid.UUID, 0)
+	staleMembersByAccount := make(map[uuid.UUID][]interface{})
 	for accountID, cmd := range commands {
 		values := cmd.Val()
 		if len(values) == 0 {
 			continue
 		}
-		clientIDText, ok := values[0].Member.(string)
-		if !ok || clientIDText == "" {
-			continue
+		seenClients := make(map[uuid.UUID]struct{}, len(values))
+		for _, value := range values {
+			clientIDText, ok := value.Member.(string)
+			if !ok || clientIDText == "" {
+				staleMembersByAccount[accountID] = append(staleMembersByAccount[accountID], value.Member)
+				continue
+			}
+			clientID, err := uuid.Parse(clientIDText)
+			if err != nil {
+				staleMembersByAccount[accountID] = append(staleMembersByAccount[accountID], clientIDText)
+				continue
+			}
+			if _, exists := seenClients[clientID]; exists {
+				continue
+			}
+			seenClients[clientID] = struct{}{}
+			result[accountID] = append(result[accountID], desktopAccountPresenceClient{
+				ClientID:             clientID,
+				OnlineSinceUnixMilli: int64(value.Score),
+			})
+			clientIDs = append(clientIDs, clientID)
 		}
-		clientID, err := uuid.Parse(clientIDText)
-		if err != nil {
-			continue
-		}
-		result[accountID] = clientID
 	}
+
+	clientIDs = uniqueUUIDs(clientIDs)
+	if len(clientIDs) == 0 {
+		pruneDesktopAccountPresenceMembers(ctx, redis, staleMembersByAccount)
+		cleanupDesktopAccountPresenceKeys(ctx, redis, uniqueIDs)
+		return nil
+	}
+
+	presencePipe := redis.Pipeline()
+	presenceCommands := make(map[uuid.UUID]*goredis.IntCmd, len(clientIDs))
+	for _, clientID := range clientIDs {
+		presenceCommands[clientID] = presencePipe.Exists(ctx, desktopClientPresenceKey(clientID))
+	}
+	if _, err := presencePipe.Exec(ctx); err != nil && err != goredis.Nil {
+		return nil
+	}
+
+	for accountID, clients := range result {
+		kept := clients[:0]
+		for _, client := range clients {
+			cmd := presenceCommands[client.ClientID]
+			if cmd != nil && cmd.Val() > 0 {
+				kept = append(kept, client)
+				continue
+			}
+			staleMembersByAccount[accountID] = append(staleMembersByAccount[accountID], client.ClientID.String())
+		}
+		if len(kept) == 0 {
+			delete(result, accountID)
+			continue
+		}
+		result[accountID] = kept
+	}
+
+	pruneDesktopAccountPresenceMembers(ctx, redis, staleMembersByAccount)
+	cleanupDesktopAccountPresenceKeys(ctx, redis, uniqueIDs)
 	return result
 }
 
+func parseDesktopPresenceUnixMilli(value string, fallback time.Time) int64 {
+	return parseDesktopPresenceTime(value, fallback).UnixMilli()
+}
+
+func parseDesktopPresenceTime(value string, fallback time.Time) time.Time {
+	if millis, err := strconv.ParseInt(value, 10, 64); err == nil && millis > 0 {
+		return time.UnixMilli(millis).UTC()
+	}
+	if parsed, err := time.Parse(time.RFC3339Nano, value); err == nil {
+		return parsed.UTC()
+	}
+	if parsed, err := time.Parse(time.RFC3339, value); err == nil {
+		return parsed.UTC()
+	}
+	return fallback.UTC()
+}
+
+func pruneDesktopAccountPresenceMembers(ctx context.Context, redis *goredis.Client, membersByAccount map[uuid.UUID][]interface{}) {
+	if redis == nil || len(membersByAccount) == 0 {
+		return
+	}
+
+	pipe := redis.Pipeline()
+	shouldExec := false
+	for accountID, members := range membersByAccount {
+		if len(members) == 0 {
+			continue
+		}
+		pipe.ZRem(ctx, desktopAccountPresenceKey(accountID), members...)
+		shouldExec = true
+	}
+	if shouldExec {
+		_, _ = pipe.Exec(ctx)
+	}
+}
+
 func uniqueUUIDs(values []uuid.UUID) []uuid.UUID {
 	if len(values) == 0 {
 		return nil