♻️ Replace normal streams with JetStream

🐛 Fix pass order didn't handled successfully

DysonNetwork.Develop/DysonNetwork.Develop.csproj

@@ -14,8 +14,6 @@
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
-        <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
-        <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
         <PackageReference Include="NodaTime.Serialization.Protobuf" Version="2.0.2" />
         <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4"/>
         <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL.NodaTime" Version="9.0.4" />

DysonNetwork.Drive/DysonNetwork.Drive.csproj

@@ -22,8 +22,6 @@
           <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
         <PackageReference Include="Minio" Version="6.0.5" />
-        <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
-        <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
         <PackageReference Include="Nerdbank.GitVersioning" Version="3.7.115">
           <PrivateAssets>all</PrivateAssets>
           <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

DysonNetwork.Drive/Startup/BroadcastEventHandler.cs

@@ -3,6 +3,8 @@ using DysonNetwork.Drive.Storage;
 using DysonNetwork.Shared.Stream;
 using Microsoft.EntityFrameworkCore;
 using NATS.Client.Core;
+using NATS.Client.JetStream.Models;
+using NATS.Net;
 
 namespace DysonNetwork.Drive.Startup;
 
@@ -14,12 +16,23 @@ public class BroadcastEventHandler(
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        await foreach (var msg in nats.SubscribeAsync<byte[]>(AccountDeletedEvent.Type, cancellationToken: stoppingToken))
+        var js = nats.CreateJetStreamContext();
+
+        await js.EnsureStreamCreated("account_events", [AccountDeletedEvent.Type]);
+        
+        var consumer = await js.CreateOrUpdateConsumerAsync("account_events",
+            new ConsumerConfig("drive_account_deleted_handler"), cancellationToken: stoppingToken);
+
+        await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
         {
             try
             {
                 var evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data);
-                if (evt == null) continue;
+                if (evt == null)
+                {
+                    await msg.AckAsync(cancellationToken: stoppingToken);
+                    continue;
+                }
 
                 logger.LogInformation("Account deleted: {AccountId}", evt.AccountId);
 
@@ -46,10 +59,13 @@ public class BroadcastEventHandler(
                     await transaction.RollbackAsync(cancellationToken: stoppingToken);
                     throw;
                 }
+
+                await msg.AckAsync(cancellationToken: stoppingToken);
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, "Error processing AccountDeleted");
+                await msg.NakAsync(cancellationToken: stoppingToken);
             }
         }
     }

DysonNetwork.Gateway/DysonNetwork.Gateway.csproj

@@ -9,8 +9,6 @@
   <ItemGroup>
     <PackageReference Include="dotnet-etcd" Version="8.0.1" />
     <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.7" />
-    <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
-    <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
     <PackageReference Include="Nerdbank.GitVersioning" Version="3.7.115">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

DysonNetwork.Pass/Account/AccountService.cs

@@ -13,6 +13,8 @@ using EFCore.BulkExtensions;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Localization;
 using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Net;
 using NodaTime;
 using OtpNet;
 using AuthService = DysonNetwork.Pass.Auth.AuthService;
@@ -189,7 +191,8 @@ public class AccountService(
         );
     }
 
-    public async Task<Account> CreateBotAccount(Account account, Guid automatedId, string? pictureId, string? backgroundId)
+    public async Task<Account> CreateBotAccount(Account account, Guid automatedId, string? pictureId,
+        string? backgroundId)
     {
         var dupeAutomateCount = await db.Accounts.Where(a => a.AutomatedId == automatedId).CountAsync();
         if (dupeAutomateCount > 0)
@@ -740,10 +743,14 @@ public class AccountService(
         db.Accounts.Remove(account);
         await db.SaveChangesAsync();
 
-        await nats.PublishAsync(AccountDeletedEvent.Type, JsonSerializer.SerializeToUtf8Bytes(new AccountDeletedEvent
+        var js = nats.CreateJetStreamContext();
+        await js.PublishAsync(
+            AccountDeletedEvent.Type,
+            GrpcTypeHelper.ConvertObjectToByteString(new AccountDeletedEvent
             {
                 AccountId = account.Id,
                 DeletedAt = SystemClock.Instance.GetCurrentInstant()
-        }));
+            }).ToByteArray()
+        );
     }
 }

DysonNetwork.Pass/DysonNetwork.Pass.csproj

@@ -14,8 +14,6 @@
             <PrivateAssets>all</PrivateAssets>
         </PackageReference>
         <PackageReference Include="Nager.Holiday" Version="1.0.1" />
-        <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
-        <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
         <PackageReference Include="Nerdbank.GitVersioning" Version="3.7.115">
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
             <PrivateAssets>all</PrivateAssets>

DysonNetwork.Pass/Startup/BroadcastEventHandler.cs

@@ -1,7 +1,10 @@
 using System.Text.Json;
 using DysonNetwork.Pass.Wallet;
+using DysonNetwork.Shared.Proto;
 using DysonNetwork.Shared.Stream;
 using NATS.Client.Core;
+using NATS.Client.JetStream.Models;
+using NATS.Net;
 
 namespace DysonNetwork.Pass.Startup;
 
@@ -13,18 +16,30 @@ public class BroadcastEventHandler(
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        await foreach (var msg in nats.SubscribeAsync<byte[]>(PaymentOrderEventBase.Type, cancellationToken: stoppingToken))
+        var js = nats.CreateJetStreamContext();
+
+        await js.EnsureStreamCreated("payment_events", [PaymentOrderEventBase.Type]);
+        
+        var consumer = await js.CreateOrUpdateConsumerAsync("payment_events", 
+            new ConsumerConfig("pass_payment_handler"), 
+            cancellationToken: stoppingToken);
+        
+        await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
         {
             PaymentOrderEvent? evt = null;
             try
             {
-                evt = JsonSerializer.Deserialize<PaymentOrderEvent>(msg.Data);
+                evt = JsonSerializer.Deserialize<PaymentOrderEvent>(msg.Data, GrpcTypeHelper.SerializerOptions);
+                
+                logger.LogInformation(
+                    "Received order event: {ProductIdentifier} {OrderId}",
+                    evt?.ProductIdentifier,
+                    evt?.OrderId
+                );
 
                 if (evt?.ProductIdentifier is null ||
                     !evt.ProductIdentifier.StartsWith(SubscriptionType.StellarProgram))
-                {
                     continue;
-                }
 
                 logger.LogInformation("Handling stellar program order: {OrderId}", evt.OrderId);
 
@@ -38,19 +53,20 @@ public class BroadcastEventHandler(
                 );
                 if (order is null)
                 {
-                    logger.LogWarning("Order with ID {OrderId} not found.", evt.OrderId);
+                    logger.LogWarning("Order with ID {OrderId} not found. Redelivering.", evt.OrderId);
-                    await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
+                    await msg.NakAsync(cancellationToken: stoppingToken);
                     continue;
                 }
 
                 await subscriptions.HandleSubscriptionOrder(order);
 
                 logger.LogInformation("Subscription for order {OrderId} handled successfully.", evt.OrderId);
+                await msg.AckAsync(cancellationToken: stoppingToken);
             }
             catch (Exception ex)
             {
-                logger.LogError(ex, "Error processing payment order event for order {OrderId}", evt?.OrderId);
+                logger.LogError(ex, "Error processing payment order event for order {OrderId}. Redelivering.", evt?.OrderId);
-                await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
+                await msg.NakAsync(cancellationToken: stoppingToken);
             }
         }
     }

DysonNetwork.Pass/Startup/ServiceCollectionExtensions.cs

@@ -211,6 +211,8 @@ public static class ServiceCollectionExtensions
         services.Configure<OidcProviderOptions>(configuration.GetSection("OidcProvider"));
         services.AddScoped<OidcProviderService>();
         
+        services.AddHostedService<BroadcastEventHandler>();
+
         return services;
     }
 }

DysonNetwork.Pass/Wallet/PaymentService.cs

@@ -7,6 +7,8 @@ using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Storage;
 using Microsoft.Extensions.Localization;
 using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Net;
 using NodaTime;
 using AccountService = DysonNetwork.Pass.Account.AccountService;
 
@@ -253,6 +255,27 @@ public class PaymentService(
             throw new InvalidOperationException("Order not found");
         }
 
+        var js = nats.CreateJetStreamContext();
+
+        if (order.Status == OrderStatus.Paid)
+        {
+            await js.PublishAsync(
+                PaymentOrderEventBase.Type,
+                GrpcTypeHelper.ConvertObjectToByteString(new PaymentOrderEvent
+                {
+                    OrderId = order.Id,
+                    WalletId = payerWallet.Id,
+                    AccountId = payerWallet.AccountId,
+                    AppIdentifier = order.AppIdentifier,
+                    ProductIdentifier = order.ProductIdentifier,
+                    Meta = order.Meta ?? [],
+                    Status = (int)order.Status,
+                }).ToByteArray()
+            );
+
+            return order;
+        }
+
         if (order.Status != OrderStatus.Unpaid)
         {
             throw new InvalidOperationException($"Order is in invalid status: {order.Status}");
@@ -282,7 +305,9 @@ public class PaymentService(
 
         await NotifyOrderPaid(order, payerWallet, order.PayeeWallet);
 
-        await nats.PublishAsync(PaymentOrderEventBase.Type, JsonSerializer.SerializeToUtf8Bytes(new PaymentOrderEvent
+        await js.PublishAsync(
+            PaymentOrderEventBase.Type,
+            GrpcTypeHelper.ConvertObjectToByteString(new PaymentOrderEvent
             {
                 OrderId = order.Id,
                 WalletId = payerWallet.Id,
@@ -291,7 +316,8 @@ public class PaymentService(
                 ProductIdentifier = order.ProductIdentifier,
                 Meta = order.Meta ?? [],
                 Status = (int)order.Status,
-        }));
+            }).ToByteArray()
+        );
 
         return order;
     }

DysonNetwork.Pusher/DysonNetwork.Pusher.csproj

@@ -19,8 +19,6 @@
           <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
           <PrivateAssets>all</PrivateAssets>
         </PackageReference>
-        <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
-        <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
         <PackageReference Include="Nerdbank.GitVersioning" Version="3.7.115">
           <PrivateAssets>all</PrivateAssets>
           <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

DysonNetwork.Pusher/Services/QueueBackgroundService.cs

@@ -3,8 +3,12 @@ using DysonNetwork.Pusher.Email;
 using DysonNetwork.Pusher.Notification;
 using DysonNetwork.Shared.Proto;
 using DysonNetwork.Shared.Registry;
+using DysonNetwork.Shared.Stream;
 using Google.Protobuf;
 using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Models;
+using NATS.Net;
 
 namespace DysonNetwork.Pusher.Services;
 
@@ -16,8 +20,8 @@ public class QueueBackgroundService(
 )
     : BackgroundService
 {
-    public const string QueueName = "pusher.queue";
+    public const string QueueName = "pusher_queue";
-    public const string QueueGroup = "pusher.workers";
+    private const string QueueGroup = "pusher_workers";
     private readonly int _consumerCount = configuration.GetValue<int?>("ConsumerCount") ?? Environment.ProcessorCount;
     private readonly List<Task> _consumerTasks = [];
 
@@ -36,11 +40,16 @@ public class QueueBackgroundService(
     private async Task RunConsumerAsync(CancellationToken stoppingToken)
     {
         logger.LogInformation("Queue consumer started");
+        var js = nats.CreateJetStreamContext();
 
-        await foreach (var msg in nats.SubscribeAsync<byte[]>(
+        await js.EnsureStreamCreated("pusher_events", [QueueName]);
-                           QueueName,
+        
-                           queueGroup: QueueGroup,
+        var consumer = await js.CreateOrUpdateConsumerAsync(
-                           cancellationToken: stoppingToken))
+            "pusher_events", 
+            new ConsumerConfig(QueueGroup), // durable consumer
+            cancellationToken: stoppingToken);
+
+        await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
         {
             try
             {
@@ -48,10 +57,12 @@ public class QueueBackgroundService(
                 if (message is not null)
                 {
                     await ProcessMessageAsync(msg, message, stoppingToken);
+                    await msg.AckAsync(cancellationToken: stoppingToken);
                 }
                 else
                 {
                     logger.LogWarning($"Invalid message format for {msg.Subject}");
+                    await msg.AckAsync(cancellationToken: stoppingToken); // Acknowledge invalid messages to avoid redelivery
                 }
             }
             catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
@@ -62,21 +73,18 @@ public class QueueBackgroundService(
             catch (Exception ex)
             {
                 logger.LogError(ex, "Error in queue consumer");
-                // Add a small delay to prevent tight error loops
+                await msg.NakAsync(cancellationToken: stoppingToken);
-                await Task.Delay(1000, stoppingToken);
             }
         }
     }
 
-    private async ValueTask ProcessMessageAsync(NatsMsg<byte[]> rawMsg, QueueMessage message,
+    private async ValueTask ProcessMessageAsync(NatsJSMsg<byte[]> rawMsg, QueueMessage message,
         CancellationToken cancellationToken)
     {
         using var scope = serviceProvider.CreateScope();
 
         logger.LogDebug("Processing message of type {MessageType}", message.Type);
 
-        try
-        {
         switch (message.Type)
         {
             case QueueMessageType.Email:
@@ -92,13 +100,6 @@ public class QueueBackgroundService(
                 break;
         }
     }
-        catch (Exception ex)
-        {
-            logger.LogError(ex, "Error processing message of type {MessageType}", message.Type);
-            // Don't rethrow to prevent the message from being retried indefinitely
-            // In a production scenario, you might want to implement a dead-letter queue
-        }
-    }
 
     private static async Task ProcessEmailMessageAsync(QueueMessage message, IServiceScope scope)
     {
@@ -126,16 +127,8 @@ public class QueueBackgroundService(
             return;
         }
 
-        try
-        {
         logger.LogDebug("Processing push notification for account {AccountId}", notification.AccountId);
         await pushService.DeliverPushNotification(notification, cancellationToken);
         logger.LogDebug("Successfully processed push notification for account {AccountId}", notification.AccountId);
     }
-        catch (Exception ex)
-        {
-            logger.LogError(ex, "Error processing push notification for account {AccountId}", notification.AccountId);
-            // Don't rethrow to prevent the message from being retried indefinitely
-        }
-    }
 }

DysonNetwork.Pusher/Services/QueueService.cs

@@ -1,6 +1,8 @@
 using System.Text.Json;
 using DysonNetwork.Shared.Proto;
 using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Net;
 
 namespace DysonNetwork.Pusher.Services;
 
@@ -20,7 +22,8 @@ public class QueueService(INatsConnection nats)
             })
         };
         var rawMessage = GrpcTypeHelper.ConvertObjectToByteString(message).ToByteArray();
-        await nats.PublishAsync(QueueBackgroundService.QueueName, rawMessage);
+        var js = nats.CreateJetStreamContext();
+        await js.PublishAsync(QueueBackgroundService.QueueName, rawMessage);
     }
 
     public async Task EnqueuePushNotification(Notification.Notification notification, Guid userId, bool isSavable = false)
@@ -35,7 +38,8 @@ public class QueueService(INatsConnection nats)
             Data = JsonSerializer.Serialize(notification)
         };
         var rawMessage = GrpcTypeHelper.ConvertObjectToByteString(message).ToByteArray();
-        await nats.PublishAsync(QueueBackgroundService.QueueName, rawMessage);
+        var js = nats.CreateJetStreamContext();
+        await js.PublishAsync(QueueBackgroundService.QueueName, rawMessage);
     }
 }
 

DysonNetwork.Shared/DysonNetwork.Shared.csproj

@@ -21,8 +21,7 @@
         <PackageReference Include="Microsoft.AspNetCore.Authentication" Version="2.3.0" />
         <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.7" />
         <PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.7" />
-        <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
+        <PackageReference Include="NATS.Net" Version="2.6.8" />
-        <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
         <PackageReference Include="NodaTime" Version="3.2.2" />
         <PackageReference Include="NodaTime.Serialization.JsonNet" Version="3.2.0" />
         <PackageReference Include="NodaTime.Serialization.Protobuf" Version="2.0.2" />

DysonNetwork.Shared/Stream/Streamer.cs

@@ -0,0 +1,23 @@
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Models;
+
+namespace DysonNetwork.Shared.Stream;
+
+public static class Streamer
+{
+    public static async Task<INatsJSStream> EnsureStreamCreated(
+        this INatsJSContext context,
+        string stream,
+        ICollection<string>? subjects
+    )
+    {
+        try
+        {
+            return await context.CreateStreamAsync(new StreamConfig(stream, subjects ?? []));
+        }
+        catch (NatsJSException)
+        {
+            return await context.GetStreamAsync(stream);
+        }
+    }
+}

DysonNetwork.Sphere/DysonNetwork.Sphere.csproj

@@ -30,8 +30,6 @@
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
-        <PackageReference Include="NATS.Client.Core" Version="2.6.8" />
-        <PackageReference Include="NATS.Client.JetStream" Version="2.6.8" />
         <PackageReference Include="Nerdbank.GitVersioning" Version="3.7.115">
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

DysonNetwork.Sphere/Startup/BroadcastEventHandler.cs

@@ -1,9 +1,12 @@
 using System.Text.Json;
 using System.Text.Json.Serialization;
+using DysonNetwork.Shared.Proto;
 using DysonNetwork.Shared.Stream;
 using DysonNetwork.Sphere.Post;
 using Microsoft.EntityFrameworkCore;
 using NATS.Client.Core;
+using NATS.Client.JetStream.Models;
+using NATS.Net;
 
 namespace DysonNetwork.Sphere.Startup;
 
@@ -16,7 +19,7 @@ public class PaymentOrderAwardMeta
 {
     [JsonPropertyName("account_id")] public Guid AccountId { get; set; }
     [JsonPropertyName("post_id")] public Guid PostId { get; set; }
-    [JsonPropertyName("amount")] public string Amount { get; set; }
+    [JsonPropertyName("amount")] public string Amount { get; set; } = null!;
     [JsonPropertyName("attitude")] public PostReactionAttitude Attitude { get; set; }
     [JsonPropertyName("message")] public string? Message { get; set; }
 }
@@ -29,14 +32,33 @@ public class BroadcastEventHandler(
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        await foreach (var msg in nats.SubscribeAsync<byte[]>(PaymentOrderEventBase.Type, cancellationToken: stoppingToken))
+        var paymentTask = HandlePaymentOrders(stoppingToken);
+        var accountTask = HandleAccountDeletions(stoppingToken);
+
+        await Task.WhenAll(paymentTask, accountTask);
+    }
+
+    private async Task HandlePaymentOrders(CancellationToken stoppingToken)
+    {
+        var js = nats.CreateJetStreamContext();
+
+        await js.EnsureStreamCreated("payment_events", [PaymentOrderEventBase.Type]);
+
+        var consumer = await js.CreateOrUpdateConsumerAsync("payment_events",
+            new ConsumerConfig("sphere_payment_handler"), cancellationToken: stoppingToken);
+
+        await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
         {
             PaymentOrderEvent? evt = null;
             try
             {
-                evt = JsonSerializer.Deserialize<PaymentOrderEvent>(msg.Data);
+                evt = JsonSerializer.Deserialize<PaymentOrderEvent>(msg.Data, GrpcTypeHelper.SerializerOptions);
                 
-                // Every order goes into the MQ is already paid, so we skipped the status validation
+                logger.LogInformation(
+                    "Received order event: {ProductIdentifier} {OrderId}",
+                    evt?.ProductIdentifier,
+                    evt?.OrderId
+                );
 
                 if (evt?.ProductIdentifier is null)
                     continue;
@@ -60,28 +82,41 @@ public class BroadcastEventHandler(
                         await ps.AwardPost(meta.PostId, meta.AccountId, amountNum, meta.Attitude, meta.Message);
 
                         logger.LogInformation("Post award for order {OrderId} handled successfully.", evt.OrderId);
-
+                        await msg.AckAsync(cancellationToken: stoppingToken);
                         break;
                     }
                     default:
-                        await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
+                        await msg.NakAsync(cancellationToken: stoppingToken);
                         break;
                 }
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, "Error processing payment order event for order {OrderId}", evt?.OrderId);
-                await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
+                await msg.NakAsync(cancellationToken: stoppingToken);
+            }
         }
     }
 
-        await foreach (var msg in nats.SubscribeAsync<byte[]>(AccountDeletedEvent.Type,
+    private async Task HandleAccountDeletions(CancellationToken stoppingToken)
-                           cancellationToken: stoppingToken))
+    {
+        var js = nats.CreateJetStreamContext();
+        
+        await js.EnsureStreamCreated("account_events", [AccountDeletedEvent.Type]);
+        
+        var consumer = await js.CreateOrUpdateConsumerAsync("account_events",
+            new ConsumerConfig("sphere_account_deleted_handler"), cancellationToken: stoppingToken);
+
+        await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
         {
             try
             {
-                var evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data);
+                var evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data, GrpcTypeHelper.SerializerOptions);
-                if (evt == null) continue;
+                if (evt == null)
+                {
+                    await msg.AckAsync(cancellationToken: stoppingToken);
+                    continue;
+                }
 
                 logger.LogInformation("Account deleted: {AccountId}", evt.AccountId);
 
@@ -120,10 +155,13 @@ public class BroadcastEventHandler(
                     await transaction.RollbackAsync(cancellationToken: stoppingToken);
                     throw;
                 }
+
+                await msg.AckAsync(cancellationToken: stoppingToken);
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, "Error processing AccountDeleted");
+                await msg.NakAsync(cancellationToken: stoppingToken);
             }
         }
     }

DysonNetwork.sln.DotSettings.user

@@ -138,6 +138,7 @@
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AStackFrameIterator_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F3bef61b8a21d4c8e96872ecdd7782fa0e55000_003F7a_003F870020d0_003FStackFrameIterator_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AStackFrameIterator_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003Fb6f0571a6bc744b0b551fd4578292582e54c00_003Fdf_003F3fcdc4d2_003FStackFrameIterator_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AStatusCodeResult_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2024_002E3_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F0b5acdd962e549369896cece0026e556214600_003F7c_003F8b7572ae_003FStatusCodeResult_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
+	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AStreamConfig_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2025_002E2_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F7140aa4493a2490fb306b1e68b5d533c98200_003Fbc_003Fccf922ff_003FStreamConfig_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AStructuredTransformer_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003Fbf3f51607a3e4e76b5b91640cd7409195c430_003F5c_003F73acd7b4_003FStructuredTransformer_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ASyndicationFeed_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F5b43b9cf654743f8b9a2eee23c625dd21dd30_003Fad_003Fd26b4d73_003FSyndicationFeed_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ASyndicationItem_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003FLibrary_003FApplication_0020Support_003FJetBrains_003FRider2025_002E1_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F5b43b9cf654743f8b9a2eee23c625dd21dd30_003Fe1_003Fb136d7be_003FSyndicationItem_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>