✨ Account service account deleted broadcast message & sphere service clean up

DysonNetwork.Sphere/Program.cs

@@ -2,6 +2,7 @@ using DysonNetwork.Shared.Auth;
 using DysonNetwork.Shared.Http;
 using DysonNetwork.Shared.PageData;
 using DysonNetwork.Shared.Registry;
+using DysonNetwork.Shared.Stream;
 using DysonNetwork.Sphere;
 using DysonNetwork.Sphere.PageData;
 using DysonNetwork.Sphere.Startup;
@@ -18,6 +19,7 @@ builder.Services.AddAppMetrics();
 
 // Add application services
 builder.Services.AddRegistryService(builder.Configuration);
+builder.Services.AddStreamConnection(builder.Configuration);
 builder.Services.AddAppServices(builder.Configuration);
 builder.Services.AddAppRateLimiting();
 builder.Services.AddAppAuthentication();

DysonNetwork.Sphere/Startup/BroadcastEventHandler.cs

@@ -0,0 +1,51 @@
+using System.Text.Json;
+using DysonNetwork.Shared.Stream;
+using Microsoft.EntityFrameworkCore;
+using NATS.Client.Core;
+
+namespace DysonNetwork.Sphere.Startup;
+
+public class BroadcastEventHandler(
+    INatsConnection nats,
+    ILogger<BroadcastEventHandler> logger,
+    AppDatabase db
+) : BackgroundService
+{
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await foreach (var msg in nats.SubscribeAsync<byte[]>("accounts.deleted", cancellationToken: stoppingToken))
+        {
+            try
+            {
+                var evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data);
+                if (evt == null) continue;
+
+                logger.LogInformation("Account deleted: {AccountId}", evt.AccountId);
+
+                await using var transaction = await db.Database.BeginTransactionAsync(cancellationToken: stoppingToken);
+                try
+                {
+                    var publishers = await db.Publishers
+                        .Where(p => p.Members.All(m => m.AccountId == evt.AccountId))
+                        .ToListAsync(cancellationToken: stoppingToken);
+
+                    foreach (var publisher in publishers)
+                        await db.Posts
+                            .Where(p => p.PublisherId == publisher.Id)
+                            .ExecuteDeleteAsync(cancellationToken: stoppingToken);
+
+                    await transaction.CommitAsync(cancellationToken: stoppingToken);
+                }
+                catch (Exception)
+                {
+                    await transaction.RollbackAsync(cancellationToken: stoppingToken);
+                    throw;
+                }
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "Error processing AccountDeleted");
+            }
+        }
+    }
+}

DysonNetwork.Sphere/Startup/ServiceCollectionExtensions.cs

@@ -73,6 +73,8 @@ public static class ServiceCollectionExtensions
             options.SupportedUICultures = supportedCultures;
         });
 
+        services.AddHostedService<BroadcastEventHandler>();
+
         return services;
     }