♻️ Extract the Storage service to DysonNetwork.Drive microservice

DysonNetwork.Drive/Handlers/ActionLogFlushHandler.cs

@@ -0,0 +1,27 @@
+
+using EFCore.BulkExtensions;
+using Quartz;
+using DysonNetwork.Sphere;
+using DysonNetwork.Common.Models;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DysonNetwork.Drive.Handlers;
+
+public class ActionLogFlushHandler(IServiceProvider serviceProvider) : IFlushHandler<ActionLog>
+{
+    public async Task FlushAsync(IReadOnlyList<ActionLog> items)
+    {
+        using var scope = serviceProvider.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDatabase>();
+
+        await db.BulkInsertAsync(items, config => config.ConflictOption = ConflictOption.Ignore);
+    }
+}
+
+public class ActionLogFlushJob(FlushBufferService fbs, ActionLogFlushHandler hdl) : IJob
+{
+    public async Task Execute(IJobExecutionContext context)
+    {
+        await fbs.FlushAsync(hdl);
+    }
+}

DysonNetwork.Drive/Handlers/LastActiveFlushHandler.cs

@@ -0,0 +1,63 @@
+using Microsoft.EntityFrameworkCore;
+using NodaTime;
+using Quartz;
+using DysonNetwork.Drive.Auth;
+using DysonNetwork.Drive.Models;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DysonNetwork.Drive.Handlers;
+
+public class LastActiveInfo
+{
+    public Session Session { get; set; } = null!;
+    public Account Account { get; set; } = null!;
+    public Instant SeenAt { get; set; }
+}
+
+public class LastActiveFlushHandler(IServiceProvider serviceProvider) : IFlushHandler<LastActiveInfo>
+{
+    public async Task FlushAsync(IReadOnlyList<LastActiveInfo> items)
+    {
+        using var scope = serviceProvider.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDatabase>();
+
+        // Remove duplicates by grouping on (sessionId, accountId), taking the most recent SeenAt
+        var distinctItems = items
+            .GroupBy(x => (SessionId: x.Session.Id, AccountId: x.Account.Id))
+            .Select(g => g.OrderByDescending(x => x.SeenAt).First())
+            .ToList();
+
+        // Build dictionaries so we can match session/account IDs to their new "last seen" timestamps
+        var sessionIdMap = distinctItems
+            .GroupBy(x => x.SessionId)
+            .ToDictionary(g => g.Key, g => g.Last().SeenAt);
+
+        var accountIdMap = distinctItems
+            .GroupBy(x => x.AccountId)
+            .ToDictionary(g => g.Key, g => g.Last().SeenAt);
+
+        // Update sessions using native EF Core ExecuteUpdateAsync
+        foreach (var kvp in sessionIdMap)
+        {
+            await db.AuthSessions
+                .Where(s => s.Id == kvp.Key)
+                .ExecuteUpdateAsync(s => s.SetProperty(x => x.LastGrantedAt, kvp.Value));
+        }
+
+        // Update account profiles using native EF Core ExecuteUpdateAsync
+        foreach (var kvp in accountIdMap)
+        {
+            await db.AccountProfiles
+                .Where(a => a.AccountId == kvp.Key)
+                .ExecuteUpdateAsync(a => a.SetProperty(x => x.LastSeenAt, kvp.Value));
+        }
+    }
+}
+
+public class LastActiveFlushJob(FlushBufferService fbs, LastActiveFlushHandler hdl) : IJob
+{
+    public async Task Execute(IJobExecutionContext context)
+    {
+        await fbs.FlushAsync(hdl);
+    }
+}

DysonNetwork.Drive/Handlers/MessageReadReceiptFlushHandler.cs

@@ -0,0 +1,35 @@
+using EFCore.BulkExtensions;
+using Microsoft.EntityFrameworkCore;
+using NodaTime;
+using Quartz;
+using DysonNetwork.Sphere;
+using DysonNetwork.Common.Models;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DysonNetwork.Drive.Handlers;
+
+public class MessageReadReceiptFlushHandler(IServiceProvider serviceProvider) : IFlushHandler<MessageReadReceipt>
+{
+    public async Task FlushAsync(IReadOnlyList<MessageReadReceipt> items)
+    {
+        var now = SystemClock.Instance.GetCurrentInstant();
+        var distinctId = items
+            .DistinctBy(x => x.SenderId)
+            .Select(x => x.SenderId)
+            .ToList();
+
+        using var scope = serviceProvider.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDatabase>();
+        await db.ChatMembers.Where(r => distinctId.Contains(r.Id))
+            .ExecuteUpdateAsync(s => s.SetProperty(m => m.LastReadAt, now)
+            );
+    }
+}
+
+public class ReadReceiptFlushJob(FlushBufferService fbs, MessageReadReceiptFlushHandler hdl) : IJob
+{
+    public async Task Execute(IJobExecutionContext context)
+    {
+        await fbs.FlushAsync(hdl);
+    }
+}

DysonNetwork.Drive/Handlers/PostViewFlushHandler.cs

@@ -0,0 +1,55 @@
+using DysonNetwork.Drive.Models;
+using Microsoft.EntityFrameworkCore;
+using NodaTime;
+using Quartz;
+using Microsoft.Extensions.DependencyInjection;
+using DysonNetwork.Drive.Services;
+
+namespace DysonNetwork.Drive.Handlers;
+
+public class PostViewFlushHandler(IServiceProvider serviceProvider) : IFlushHandler<PostViewInfo>
+{
+    public async Task FlushAsync(IReadOnlyList<PostViewInfo> items)
+    {
+        using var scope = serviceProvider.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDatabase>();
+        var cache = scope.ServiceProvider.GetRequiredService<ICacheService>();
+
+        // Group views by post
+        var postViews = items
+            .GroupBy(x => x.PostId)
+            .ToDictionary(g => g.Key, g => g.ToList());
+
+        // Calculate total views and unique views per post
+        foreach (var postId in postViews.Keys)
+        {
+            // Calculate unique views by distinct viewer IDs (not null)
+            var uniqueViews = postViews[postId]
+                .Where(v => !string.IsNullOrEmpty(v.ViewerId))
+                .Select(v => v.ViewerId)
+                .Distinct()
+                .Count();
+
+            // Total views is just the count of all items for this post
+            var totalViews = postViews[postId].Count;
+
+            // Update the post in the database
+            await db.Posts
+                .Where(p => p.Id == postId)
+                .ExecuteUpdateAsync(p => p
+                    .SetProperty(x => x.ViewsTotal, x => x.ViewsTotal + totalViews)
+                    .SetProperty(x => x.ViewsUnique, x => x.ViewsUnique + uniqueViews));
+
+            // Invalidate any cache entries for this post
+            await cache.RemoveAsync($"post:{postId}");
+        }
+    }
+}
+
+public class PostViewFlushJob(FlushBufferService fbs, PostViewFlushHandler hdl) : IJob
+{
+    public async Task Execute(IJobExecutionContext context)
+    {
+        await fbs.FlushAsync(hdl);
+    }
+}