♻️ Refactored presence update logic

DysonNetwork.Pass/Account/AccountEventService.cs

@@ -7,7 +7,6 @@ using DysonNetwork.Shared.Stream;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Localization;
 using NATS.Client.Core;
-using NATS.Net;
 using NodaTime;
 using NodaTime.Extensions;
 
@@ -28,7 +27,7 @@ public class AccountEventService(
     private const string StatusCacheKey = "account:status:";
     private const string ActivityCacheKey = "account:activities:";
 
-    private async Task<bool> GetAccountIsConnected(Guid userId)
+    public async Task<bool> GetAccountIsConnected(Guid userId)
     {
         var resp = await pusher.GetWebsocketConnectionStatusAsync(
             new GetWebsocketConnectionStatusRequest { UserId = userId.ToString() }
@@ -36,6 +35,16 @@ public class AccountEventService(
         return resp.IsConnected;
     }
 
+    public async Task<Dictionary<string, bool>> GetAccountIsConnectedBatch(List<Guid> userIds)
+    {
+        var req = new GetWebsocketConnectionStatusBatchRequest();
+        req.UsersId.AddRange(userIds.Select(u => u.ToString()));
+        var resp = await pusher.GetWebsocketConnectionStatusBatchAsync(
+            req
+        );
+        return resp.IsConnected.ToDictionary();
+    }
+
     public void PurgeStatusCache(Guid userId)
     {
         var cacheKey = $"{StatusCacheKey}{userId}";
@@ -531,9 +540,9 @@ public class AccountEventService(
     )
     {
         var now = SystemClock.Instance.GetCurrentInstant();
-        var activity = await db.PresenceActivities.FirstOrDefaultAsync(
-                e => e.ManualId == manualId && e.AccountId == userId && e.LeaseExpiresAt > now && e.DeletedAt == null
-            );
+        var activity = await db.PresenceActivities.FirstOrDefaultAsync(e =>
+            e.ManualId == manualId && e.AccountId == userId && e.LeaseExpiresAt > now && e.DeletedAt == null
+        );
         if (activity == null)
             return null;
 
@@ -558,8 +567,8 @@ public class AccountEventService(
     public async Task<bool> DeleteActivityByManualId(string manualId, Guid userId)
     {
         var now = SystemClock.Instance.GetCurrentInstant();
-        var activity = await db.PresenceActivities.FirstOrDefaultAsync(
-            e => e.ManualId == manualId && e.AccountId == userId && e.LeaseExpiresAt > now && e.DeletedAt == null
+        var activity = await db.PresenceActivities.FirstOrDefaultAsync(e =>
+            e.ManualId == manualId && e.AccountId == userId && e.LeaseExpiresAt > now && e.DeletedAt == null
         );
         if (activity == null) return false;
         if (activity.LeaseExpiresAt <= now)
@@ -600,4 +609,16 @@ public class AccountEventService(
         PurgeActivityCache(activity.AccountId);
         return true;
     }
+
+    /// <summary>
+    /// Gets all user IDs that have Spotify connections
+    /// </summary>
+    public async Task<List<Guid>> GetSpotifyConnectedUsersAsync()
+    {
+        return await db.AccountConnections
+            .Where(c => c.Provider == "spotify" && c.AccessToken != null && c.RefreshToken != null)
+            .Select(c => c.AccountId)
+            .Distinct()
+            .ToListAsync();
+    }
 }

DysonNetwork.Pass/Account/Presences/IPresenceService.cs

@@ -0,0 +1,21 @@
+using DysonNetwork.Shared.Models;
+
+namespace DysonNetwork.Pass.Account.Presences;
+
+/// <summary>
+/// Interface for presence services that can update user presence activities
+/// </summary>
+public interface IPresenceService
+{
+    /// <summary>
+    /// The unique identifier for this presence service (e.g., "spotify", "discord")
+    /// </summary>
+    string ServiceId { get; }
+
+    /// <summary>
+    /// Updates presence activities for the specified users
+    /// </summary>
+    /// <param name="userIds">The user IDs to update presence for</param>
+    /// <returns>A task representing the asynchronous operation</returns>
+    Task UpdatePresencesAsync(IEnumerable<Guid> userIds);
+}

DysonNetwork.Pass/Account/Presences/PresenceUpdateJob.cs

@@ -0,0 +1,112 @@
+using Quartz;
+
+namespace DysonNetwork.Pass.Account.Presences;
+
+public class PresenceUpdateJob(
+    IEnumerable<IPresenceService> presenceServices,
+    AccountEventService accountEventService,
+    ILogger<PresenceUpdateJob> logger
+) : IJob
+{
+    public async Task Execute(IJobExecutionContext context)
+    {
+        // Get the stage parameter from the job data
+        var stageString = context.JobDetail.JobDataMap.GetString("stage");
+        if (!Enum.TryParse<PresenceUpdateStage>(stageString, out var stage))
+        {
+            logger.LogError("Invalid or missing stage parameter: {Stage}", stageString);
+            return;
+        }
+
+        logger.LogInformation("Starting presence updates for stage: {Stage}", stage);
+
+        try
+        {
+            // Get users to update based on the stage
+            var userIds = await GetUsersForStageAsync(stage);
+
+            if (userIds.Count == 0)
+            {
+                logger.LogInformation("No users found for stage {Stage}", stage);
+                return;
+            }
+
+            logger.LogInformation("Found {UserCount} users for stage {Stage}", userIds.Count, stage);
+
+            // Update presence for each service
+            foreach (var presenceService in presenceServices)
+            {
+                try
+                {
+                    await presenceService.UpdatePresencesAsync(userIds);
+                    logger.LogInformation("Updated {ServiceId} presences for {UserCount} users in stage {Stage}",
+                        presenceService.ServiceId, userIds.Count, stage);
+                }
+                catch (Exception ex)
+                {
+                    logger.LogError(ex, "Error updating {ServiceId} presences for stage {Stage}",
+                        presenceService.ServiceId, stage);
+                }
+            }
+
+            logger.LogInformation("Presence updates completed for stage {Stage}", stage);
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Error occurred during presence updates for stage {Stage}", stage);
+        }
+    }
+
+    private async Task<List<Guid>> GetUsersForStageAsync(PresenceUpdateStage stage)
+    {
+        // Get all users with presence connections
+        var allUserIds = await GetAllUsersWithPresenceConnectionsAsync();
+
+        if (!allUserIds.Any())
+        {
+            return new List<Guid>();
+        }
+
+        // Batch fetch online status for all users
+        var onlineStatuses = await accountEventService.GetAccountIsConnectedBatch(allUserIds);
+
+        var filteredUserIds = new List<Guid>();
+
+        foreach (var userId in allUserIds)
+        {
+            var userIdString = userId.ToString();
+            var isOnline = onlineStatuses.GetValueOrDefault(userIdString, false);
+            var activeActivities = await accountEventService.GetActiveActivities(userId);
+            var hasActivePresence = activeActivities.Any();
+
+            var shouldInclude = stage switch
+            {
+                PresenceUpdateStage.Active => isOnline && hasActivePresence,
+                PresenceUpdateStage.Maybe => isOnline && !hasActivePresence,
+                PresenceUpdateStage.Cold => !isOnline,
+                _ => false
+            };
+
+            if (shouldInclude)
+            {
+                filteredUserIds.Add(userId);
+            }
+        }
+
+        return filteredUserIds;
+    }
+
+    private async Task<List<Guid>> GetAllUsersWithPresenceConnectionsAsync()
+    {
+        // This method should return all users who have connections to any presence service
+        // For now, we'll focus on Spotify users, but this should be extended to include all presence services
+
+        // In a more complete implementation, you might want to query all presence services
+        // to get users with connections to any of them
+
+        // For simplicity, we'll return users with Spotify connections
+        // This should be made more generic in the future
+        var spotifyUsers = await accountEventService.GetSpotifyConnectedUsersAsync();
+        return spotifyUsers;
+    }
+}

DysonNetwork.Pass/Account/Presences/PresenceUpdateStage.cs

@@ -0,0 +1,19 @@
+namespace DysonNetwork.Pass.Account.Presences;
+
+public enum PresenceUpdateStage
+{
+    /// <summary>
+    /// Active users - online and have active presence activities
+    /// </summary>
+    Active,
+
+    /// <summary>
+    /// Maybe active users - online but no active presence activities
+    /// </summary>
+    Maybe,
+
+    /// <summary>
+    /// Cold users - offline users
+    /// </summary>
+    Cold
+}

DysonNetwork.Pass/Account/Presences/SpotifyPresenceService.cs

@@ -10,15 +10,17 @@ public class SpotifyPresenceService(
     Auth.OpenId.SpotifyOidcService spotifyService,
     AccountEventService accountEventService,
     ILogger<SpotifyPresenceService> logger
-)
+) : IPresenceService
 {
-    /// <summary>
-    /// Updates presence activities for users who have Spotify connections and are currently playing music
-    /// </summary>
-    public async Task UpdateAllSpotifyPresencesAsync()
+    /// <inheritdoc />
+    public string ServiceId => "spotify";
+
+    /// <inheritdoc />
+    public async Task UpdatePresencesAsync(IEnumerable<Guid> userIds)
     {
+        var userIdList = userIds.ToList();
         var userConnections = await db.AccountConnections
-            .Where(c => c.Provider == "spotify" && c.AccessToken != null && c.RefreshToken != null)
+            .Where(c => userIdList.Contains(c.AccountId) && c.Provider == "spotify" && c.AccessToken != null && c.RefreshToken != null)
             .Include(c => c.Account)
             .ToListAsync();
 

DysonNetwork.Pass/Account/Presences/SpotifyPresenceUpdateJob.cs

@@ -1,21 +0,0 @@
-using Quartz;
-
-namespace DysonNetwork.Pass.Account.Presences;
-
-public class SpotifyPresenceUpdateJob(SpotifyPresenceService spotifyPresenceService, ILogger<SpotifyPresenceUpdateJob> logger) : IJob
-{
-    public async Task Execute(IJobExecutionContext context)
-    {
-        logger.LogInformation("Starting Spotify presence updates...");
-
-        try
-        {
-            await spotifyPresenceService.UpdateAllSpotifyPresencesAsync();
-            logger.LogInformation("Spotify presence updates completed successfully.");
-        }
-        catch (Exception ex)
-        {
-            logger.LogError(ex, "Error occurred during Spotify presence updates.");
-        }
-    }
-}

DysonNetwork.Pass/Startup/ScheduledJobsConfiguration.cs

@@ -84,14 +84,38 @@ public static class ScheduledJobsConfiguration
                 .WithIdentity("SocialCreditValidationTrigger")
                 .WithCronSchedule("0 0 0 * * ?"));
 
-            var spotifyPresenceUpdateJob = new JobKey("SpotifyPresenceUpdate");
-            q.AddJob<SpotifyPresenceUpdateJob>(opts => opts.WithIdentity(spotifyPresenceUpdateJob));
+            // Presence update jobs for different user stages
+            var activePresenceUpdateJob = new JobKey("ActivePresenceUpdate");
+            q.AddJob<PresenceUpdateJob>(opts => opts.WithIdentity(activePresenceUpdateJob));
             q.AddTrigger(opts => opts
-                .ForJob(spotifyPresenceUpdateJob)
-                .WithIdentity("SpotifyPresenceUpdateTrigger")
+                .ForJob(activePresenceUpdateJob)
+                .WithIdentity("ActivePresenceUpdateTrigger")
                 .WithSimpleSchedule(o => o
-                    .WithIntervalInMinutes(2)
+                    .WithIntervalInMinutes(1)
                     .RepeatForever())
+                .UsingJobData("stage", nameof(PresenceUpdateStage.Active))
+            );
+
+            var maybePresenceUpdateJob = new JobKey("MaybePresenceUpdate");
+            q.AddJob<PresenceUpdateJob>(opts => opts.WithIdentity(maybePresenceUpdateJob));
+            q.AddTrigger(opts => opts
+                .ForJob(maybePresenceUpdateJob)
+                .WithIdentity("MaybePresenceUpdateTrigger")
+                .WithSimpleSchedule(o => o
+                    .WithIntervalInMinutes(3)
+                    .RepeatForever())
+                .UsingJobData("stage", nameof(PresenceUpdateStage.Maybe))
+            );
+
+            var coldPresenceUpdateJob = new JobKey("ColdPresenceUpdate");
+            q.AddJob<PresenceUpdateJob>(opts => opts.WithIdentity(coldPresenceUpdateJob));
+            q.AddTrigger(opts => opts
+                .ForJob(coldPresenceUpdateJob)
+                .WithIdentity("ColdPresenceUpdateTrigger")
+                .WithSimpleSchedule(o => o
+                    .WithIntervalInMinutes(10)
+                    .RepeatForever())
+                .UsingJobData("stage", nameof(PresenceUpdateStage.Cold))
             );
         });
         services.AddQuartzHostedService(q => q.WaitForJobsToComplete = true);

DysonNetwork.Pass/Startup/ServiceCollectionExtensions.cs

@@ -10,12 +10,13 @@ using NodaTime;
 using NodaTime.Serialization.SystemTextJson;
 using System.Text.Json;
 using System.Text.Json.Serialization;
-using System.Threading.RateLimiting;
+using DysonNetwork.Pass.Account.Presences;
 using DysonNetwork.Pass.Auth.OidcProvider.Options;
 using DysonNetwork.Pass.Auth.OidcProvider.Services;
 using DysonNetwork.Pass.Credit;
 using DysonNetwork.Pass.Handlers;
 using DysonNetwork.Pass.Leveling;
+using DysonNetwork.Pass.Lotteries;
 using DysonNetwork.Pass.Mailer;
 using DysonNetwork.Pass.Realm;
 using DysonNetwork.Pass.Safety;
@@ -144,7 +145,6 @@ public static class ServiceCollectionExtensions
         services.AddScoped<ActionLogService>();
         services.AddScoped<RelationshipService>();
         services.AddScoped<MagicSpellService>();
-        services.AddScoped<DysonNetwork.Pass.Account.Presences.SpotifyPresenceService>();
         services.AddScoped<AuthService>();
         services.AddScoped<TokenAuthService>();
         services.AddScoped<AccountUsernameService>();
@@ -156,7 +156,10 @@ public static class ServiceCollectionExtensions
         services.AddScoped<SocialCreditService>();
         services.AddScoped<ExperienceService>();
         services.AddScoped<RealmService>();
-        services.AddScoped<Lotteries.LotteryService>();
+        services.AddScoped<LotteryService>();
+        
+        services.AddScoped<SpotifyPresenceService>();
+        services.AddScoped<IPresenceService, SpotifyPresenceService>();
         
         services.Configure<OidcProviderOptions>(configuration.GetSection("OidcProvider"));
         services.AddScoped<OidcProviderService>();

DysonNetwork.Ring/Services/PusherServiceGrpc.cs

@@ -2,6 +2,7 @@ using DysonNetwork.Ring.Connection;
 using DysonNetwork.Ring.Notification;
 using DysonNetwork.Shared.Models;
 using DysonNetwork.Shared.Proto;
+using Google.Protobuf.Collections;
 using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
 
@@ -44,7 +45,7 @@ public class RingServiceGrpc(
     }
 
     public override Task<Empty> PushWebSocketPacketToDevice(PushWebSocketPacketToDeviceRequest request,
-            ServerCallContext context)
+        ServerCallContext context)
     {
         var packet = Shared.Models.WebSocketPacket.FromProtoValue(request.Packet);
 
@@ -139,10 +140,24 @@ public class RingServiceGrpc(
         {
             GetWebsocketConnectionStatusRequest.IdOneofCase.DeviceId =>
                 WebSocketService.GetDeviceIsConnected(request.DeviceId),
-            GetWebsocketConnectionStatusRequest.IdOneofCase.UserId => WebSocketService.GetAccountIsConnected(Guid.Parse(request.UserId)),
+            GetWebsocketConnectionStatusRequest.IdOneofCase.UserId => WebSocketService.GetAccountIsConnected(
+                Guid.Parse(request.UserId)),
             _ => false
         };
 
         return Task.FromResult(new GetWebsocketConnectionStatusResponse { IsConnected = isConnected });
     }
-}
+
+    public override Task<GetWebsocketConnectionStatusBatchResponse> GetWebsocketConnectionStatusBatch(
+        GetWebsocketConnectionStatusBatchRequest request, ServerCallContext context)
+    {
+        var resp = new GetWebsocketConnectionStatusBatchResponse();
+        foreach (var id in request.UsersId)
+        {
+            var gid = Guid.Parse(id);
+            resp.IsConnected[id] = WebSocketService.GetAccountIsConnected(gid);
+        }
+
+        return Task.FromResult(resp);
+    }
+}

DysonNetwork.Shared/Proto/ring.proto

@@ -12,117 +12,126 @@ import "account.proto";
 
 // RingService provides methods to send various types of notifications.
 service RingService {
-  // Sends an email.
-  rpc SendEmail(SendEmailRequest) returns (google.protobuf.Empty) {}
+    // Sends an email.
+    rpc SendEmail(SendEmailRequest) returns (google.protobuf.Empty) {}
 
-  // Pushes a packet to a user via WebSocket.
-  rpc PushWebSocketPacket(PushWebSocketPacketRequest) returns (google.protobuf.Empty) {}
+    // Pushes a packet to a user via WebSocket.
+    rpc PushWebSocketPacket(PushWebSocketPacketRequest) returns (google.protobuf.Empty) {}
 
-  // Pushes a packet to a list of users via WebSocket.
-  rpc PushWebSocketPacketToUsers(PushWebSocketPacketToUsersRequest) returns (google.protobuf.Empty) {}
+    // Pushes a packet to a list of users via WebSocket.
+    rpc PushWebSocketPacketToUsers(PushWebSocketPacketToUsersRequest) returns (google.protobuf.Empty) {}
 
-  // Pushes a packet to a device via WebSocket.
-  rpc PushWebSocketPacketToDevice(PushWebSocketPacketToDeviceRequest) returns (google.protobuf.Empty) {}
+    // Pushes a packet to a device via WebSocket.
+    rpc PushWebSocketPacketToDevice(PushWebSocketPacketToDeviceRequest) returns (google.protobuf.Empty) {}
 
-  // Pushes a packet to a list of devices via WebSocket.
-  rpc PushWebSocketPacketToDevices(PushWebSocketPacketToDevicesRequest) returns (google.protobuf.Empty) {}
+    // Pushes a packet to a list of devices via WebSocket.
+    rpc PushWebSocketPacketToDevices(PushWebSocketPacketToDevicesRequest) returns (google.protobuf.Empty) {}
 
-  // Sends a push notification to a user.
-  rpc SendPushNotificationToUser(SendPushNotificationToUserRequest) returns (google.protobuf.Empty) {}
+    // Sends a push notification to a user.
+    rpc SendPushNotificationToUser(SendPushNotificationToUserRequest) returns (google.protobuf.Empty) {}
 
-  // Sends a push notification to a list of users.
-  rpc SendPushNotificationToUsers(SendPushNotificationToUsersRequest) returns (google.protobuf.Empty) {}
+    // Sends a push notification to a list of users.
+    rpc SendPushNotificationToUsers(SendPushNotificationToUsersRequest) returns (google.protobuf.Empty) {}
 
-  // Unsubscribes a device from push notifications.
-  rpc UnsubscribePushNotifications(UnsubscribePushNotificationsRequest) returns (google.protobuf.Empty) {}
+    // Unsubscribes a device from push notifications.
+    rpc UnsubscribePushNotifications(UnsubscribePushNotificationsRequest) returns (google.protobuf.Empty) {}
 
-  // Gets the WebSocket connection status for a device or user.
-  rpc GetWebsocketConnectionStatus(GetWebsocketConnectionStatusRequest) returns (GetWebsocketConnectionStatusResponse) {}
+    // Gets the WebSocket connection status for a device or user.
+    rpc GetWebsocketConnectionStatus(GetWebsocketConnectionStatusRequest) returns (GetWebsocketConnectionStatusResponse) {}
+
+    rpc GetWebsocketConnectionStatusBatch(GetWebsocketConnectionStatusBatchRequest) returns (GetWebsocketConnectionStatusBatchResponse) {}
 }
 
 // Represents an email message.
 message EmailMessage {
-  string to_name = 1;
-  string to_address = 2;
-  string subject = 3;
-  string body = 4;
+    string to_name = 1;
+    string to_address = 2;
+    string subject = 3;
+    string body = 4;
 }
 
 message SendEmailRequest {
-  EmailMessage email = 1;
+    EmailMessage email = 1;
 }
 
 // Represents a WebSocket packet.
 message WebSocketPacket {
-  string type = 1;
-  bytes data = 2;
-  google.protobuf.StringValue error_message = 3;
+    string type = 1;
+    bytes data = 2;
+    google.protobuf.StringValue error_message = 3;
 }
 
 message PushWebSocketPacketRequest {
-  string user_id = 1;
-  WebSocketPacket packet = 2;
+    string user_id = 1;
+    WebSocketPacket packet = 2;
 }
 
 message PushWebSocketPacketToUsersRequest {
-  repeated string user_ids = 1;
-  WebSocketPacket packet = 2;
+    repeated string user_ids = 1;
+    WebSocketPacket packet = 2;
 }
 
 message PushWebSocketPacketToDeviceRequest {
-  string device_id = 1;
-  WebSocketPacket packet = 2;
+    string device_id = 1;
+    WebSocketPacket packet = 2;
 }
 
 message PushWebSocketPacketToDevicesRequest {
-  repeated string device_ids = 1;
-  WebSocketPacket packet = 2;
+    repeated string device_ids = 1;
+    WebSocketPacket packet = 2;
 }
 
 // Represents a push notification.
 message PushNotification {
-  string topic = 1;
-  string title = 2;
-  string subtitle = 3;
-  string body = 4;
-  optional bytes meta = 5;
-  optional string action_uri = 6;
-  bool is_silent = 7;
-  bool is_savable = 8;
+    string topic = 1;
+    string title = 2;
+    string subtitle = 3;
+    string body = 4;
+    optional bytes meta = 5;
+    optional string action_uri = 6;
+    bool is_silent = 7;
+    bool is_savable = 8;
 }
 
 message SendPushNotificationToUserRequest {
-  string user_id = 1;
-  PushNotification notification = 2;
+    string user_id = 1;
+    PushNotification notification = 2;
 }
 
 message SendPushNotificationToUsersRequest {
-  repeated string user_ids = 1;
-  PushNotification notification = 2;
+    repeated string user_ids = 1;
+    PushNotification notification = 2;
 }
 
 message UnsubscribePushNotificationsRequest {
-  string device_id = 1;
+    string device_id = 1;
 }
 
 message GetWebsocketConnectionStatusRequest {
-  oneof id {
-    string device_id = 1;
-    string user_id = 2;
-  }
+    oneof id {
+        string device_id = 1;
+        string user_id = 2;
+    }
 }
 
 message GetWebsocketConnectionStatusResponse {
-  bool is_connected = 1;
+    bool is_connected = 1;
 }
 
+message GetWebsocketConnectionStatusBatchRequest {
+    repeated string users_id = 1;
+}
+
+message GetWebsocketConnectionStatusBatchResponse {
+    map<string, bool> is_connected = 1;
+}
 
 service RingHandlerService {
-  rpc ReceiveWebSocketPacket(ReceiveWebSocketPacketRequest) returns (google.protobuf.Empty) {}
+    rpc ReceiveWebSocketPacket(ReceiveWebSocketPacketRequest) returns (google.protobuf.Empty) {}
 }
 
 message ReceiveWebSocketPacketRequest {
-  WebSocketPacket packet = 1;
-  Account account = 2;
-  string device_id = 3;
+    WebSocketPacket packet = 1;
+    Account account = 2;
+    string device_id = 3;
 }