SolarNetwork/Swarm
2
3
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

⏪ Remove jetstream

Browse Source
This commit is contained in:
LittleSheep
2025-09-11 19:14:30 +08:00
parent dbc2c54ab0
commit 9d282b26f3
3 changed files with 40 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
DysonNetwork.Drive/Startup/BroadcastEventHandler.cs
View File

@@ -3,8 +3,6 @@ using DysonNetwork.Drive.Storage;
using DysonNetwork.Shared.Stream; using DysonNetwork.Shared.Stream;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using NATS.Client.Core; using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
namespace DysonNetwork.Drive.Startup; namespace DysonNetwork.Drive.Startup;
@@ -16,23 +14,14 @@ public class BroadcastEventHandler(
{ {
protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{ {
var js = new NatsJSContext(nats); await foreach (var msg in nats.SubscribeAsync<byte[]>(AccountDeletedEvent.Type, cancellationToken: stoppingToken))
var stream = await js.GetStreamAsync(AccountDeletedEvent.Type, cancellationToken: stoppingToken);
var consumer = await stream.CreateOrUpdateConsumerAsync(new ConsumerConfig("Dy_Drive_AccountDeleted"), cancellationToken: stoppingToken);
await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
{ {
AccountDeletedEvent? evt = null;
try try
{ {
evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data); var evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data);
if (evt == null) if (evt == null) continue;
{
await msg.AckAsync(cancellationToken: stoppingToken);
continue;
}
logger.LogInformation("Processing account deletion for: {AccountId}", evt.AccountId); logger.LogInformation("Account deleted: {AccountId}", evt.AccountId);
using var scope = serviceProvider.CreateScope(); using var scope = serviceProvider.CreateScope();
var fs = scope.ServiceProvider.GetRequiredService<FileService>(); var fs = scope.ServiceProvider.GetRequiredService<FileService>();
@@ -45,30 +34,22 @@ public class BroadcastEventHandler(
.Where(p => p.AccountId == evt.AccountId) .Where(p => p.AccountId == evt.AccountId)
.ToListAsync(cancellationToken: stoppingToken); .ToListAsync(cancellationToken: stoppingToken);
if (files.Any()) await fs.DeleteFileDataBatchAsync(files);
{ await db.Files
await fs.DeleteFileDataBatchAsync(files); .Where(p => p.AccountId == evt.AccountId)
await db.Files .ExecuteDeleteAsync(cancellationToken: stoppingToken);
.Where(p => p.AccountId == evt.AccountId)
.ExecuteDeleteAsync(cancellationToken: stoppingToken);
}
await transaction.CommitAsync(cancellationToken: stoppingToken); await transaction.CommitAsync(cancellationToken: stoppingToken);
await msg.AckAsync(cancellationToken: stoppingToken);
logger.LogInformation("Account deletion for {AccountId} processed successfully.", evt.AccountId);
} }
catch (Exception ex) catch (Exception)
{ {
logger.LogError(ex, "Error during transaction for account deletion {AccountId}, rolling back.", evt.AccountId); await transaction.RollbackAsync(cancellationToken: stoppingToken);
await transaction.RollbackAsync(CancellationToken.None); throw;
throw; // Let outer catch handle Nak
} }
} }
catch (Exception ex) catch (Exception ex)
{ {
logger.LogError(ex, "Failed to process account deletion for {AccountId}, will retry.", evt?.AccountId); logger.LogError(ex, "Error processing AccountDeleted");
await msg.NakAsync(delay: TimeSpan.FromSeconds(30), cancellationToken: stoppingToken);
} }
} }
} }

14
DysonNetwork.Pass/Startup/BroadcastEventHandler.cs
View File

@@ -2,8 +2,6 @@ using System.Text.Json;
using DysonNetwork.Pass.Wallet; using DysonNetwork.Pass.Wallet;
using DysonNetwork.Shared.Stream; using DysonNetwork.Shared.Stream;
using NATS.Client.Core; using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
namespace DysonNetwork.Pass.Startup; namespace DysonNetwork.Pass.Startup;
@@ -15,11 +13,7 @@ public class BroadcastEventHandler(
{ {
protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{ {
var js = new NatsJSContext(nats); await foreach (var msg in nats.SubscribeAsync<byte[]>(PaymentOrderEventBase.Type, cancellationToken: stoppingToken))
var stream = await js.GetStreamAsync(PaymentOrderEventBase.Type, cancellationToken: stoppingToken);
var consumer = await stream.CreateOrUpdateConsumerAsync(new ConsumerConfig("Dy_Pass_Stellar"), cancellationToken: stoppingToken);
await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
{ {
PaymentOrderEvent? evt = null; PaymentOrderEvent? evt = null;
try try
@@ -29,7 +23,6 @@ public class BroadcastEventHandler(
if (evt?.ProductIdentifier is null || if (evt?.ProductIdentifier is null ||
!evt.ProductIdentifier.StartsWith(SubscriptionType.StellarProgram)) !evt.ProductIdentifier.StartsWith(SubscriptionType.StellarProgram))
{ {
await msg.AckAsync(cancellationToken: stoppingToken);
continue; continue;
} }
@@ -46,19 +39,18 @@ public class BroadcastEventHandler(
if (order is null) if (order is null)
{ {
logger.LogWarning("Order with ID {OrderId} not found.", evt.OrderId); logger.LogWarning("Order with ID {OrderId} not found.", evt.OrderId);
await msg.NakAsync(delay: TimeSpan.FromSeconds(30), cancellationToken: stoppingToken); await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
continue; continue;
} }
await subscriptions.HandleSubscriptionOrder(order); await subscriptions.HandleSubscriptionOrder(order);
logger.LogInformation("Subscription for order {OrderId} handled successfully.", evt.OrderId); logger.LogInformation("Subscription for order {OrderId} handled successfully.", evt.OrderId);
await msg.AckAsync(cancellationToken: stoppingToken);
} }
catch (Exception ex) 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}", evt?.OrderId);
await msg.NakAsync(delay: TimeSpan.FromSeconds(30), cancellationToken: stoppingToken); await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
} }
} }
} }

102
DysonNetwork.Sphere/Startup/BroadcastEventHandler.cs
View File

@@ -4,8 +4,6 @@ using DysonNetwork.Shared.Stream;
using DysonNetwork.Sphere.Post; using DysonNetwork.Sphere.Post;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using NATS.Client.Core; using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
namespace DysonNetwork.Sphere.Startup; namespace DysonNetwork.Sphere.Startup;
@@ -31,38 +29,17 @@ public class BroadcastEventHandler(
{ {
protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{ {
try await foreach (var msg in nats.SubscribeAsync<byte[]>(PaymentOrderEventBase.Type, cancellationToken: stoppingToken))
{
var paymentTask = ProcessPaymentOrdersAsync(stoppingToken);
var accountTask = ProcessAccountDeletionsAsync(stoppingToken);
await Task.WhenAll(paymentTask, accountTask);
}
catch (Exception ex)
{
logger.LogError(ex, "BroadcastEventHandler stopped due to an unhandled exception.");
}
}
private async Task ProcessPaymentOrdersAsync(CancellationToken stoppingToken)
{
var js = new NatsJSContext(nats);
var stream = await js.GetStreamAsync(PaymentOrderEventBase.Type, cancellationToken: stoppingToken);
var consumer = await stream.CreateOrUpdateConsumerAsync(new ConsumerConfig("Dy_Sphere_PaymentOrder"),
cancellationToken: stoppingToken);
await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
{ {
PaymentOrderEvent? evt = null; PaymentOrderEvent? evt = null;
try try
{ {
evt = JsonSerializer.Deserialize<PaymentOrderEvent>(msg.Data); evt = JsonSerializer.Deserialize<PaymentOrderEvent>(msg.Data);
// Every order goes into the MQ is already paid, so we skipped the status validation
if (evt?.ProductIdentifier is null) if (evt?.ProductIdentifier is null)
{
await msg.AckAsync(cancellationToken: stoppingToken);
continue; continue;
}
switch (evt.ProductIdentifier) switch (evt.ProductIdentifier)
{ {
@@ -70,9 +47,9 @@ public class BroadcastEventHandler(
{ {
var awardEvt = JsonSerializer.Deserialize<PaymentOrderAwardEvent>(msg.Data); var awardEvt = JsonSerializer.Deserialize<PaymentOrderAwardEvent>(msg.Data);
if (awardEvt?.Meta == null) throw new ArgumentNullException(nameof(awardEvt)); if (awardEvt?.Meta == null) throw new ArgumentNullException(nameof(awardEvt));
var meta = awardEvt.Meta; var meta = awardEvt.Meta;
logger.LogInformation("Handling post award order: {OrderId}", evt.OrderId); logger.LogInformation("Handling post award order: {OrderId}", evt.OrderId);
await using var scope = serviceProvider.CreateAsyncScope(); await using var scope = serviceProvider.CreateAsyncScope();
@@ -84,45 +61,29 @@ public class BroadcastEventHandler(
logger.LogInformation("Post award for order {OrderId} handled successfully.", evt.OrderId); logger.LogInformation("Post award for order {OrderId} handled successfully.", evt.OrderId);
await msg.AckAsync(cancellationToken: stoppingToken);
break; break;
} }
default: default:
// Not for us, acknowledge and ignore. await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
await msg.AckAsync(cancellationToken: stoppingToken);
break; break;
} }
} }
catch (Exception ex) catch (Exception ex)
{ {
logger.LogError(ex, "Error processing payment order event for order {OrderId}, will retry.", logger.LogError(ex, "Error processing payment order event for order {OrderId}", evt?.OrderId);
evt?.OrderId); await nats.PublishAsync(PaymentOrderEventBase.Type, msg.Data, cancellationToken: stoppingToken);
await msg.NakAsync(delay: TimeSpan.FromSeconds(30), cancellationToken: stoppingToken);
} }
} }
}
private async Task ProcessAccountDeletionsAsync(CancellationToken stoppingToken) await foreach (var msg in nats.SubscribeAsync<byte[]>(AccountDeletedEvent.Type,
{ cancellationToken: stoppingToken))
var js = new NatsJSContext(nats);
var stream = await js.GetStreamAsync(AccountDeletedEvent.Type, cancellationToken: stoppingToken);
var consumer =
await stream.CreateOrUpdateConsumerAsync(new ConsumerConfig("Dy_Sphere_AccountDeleted"),
cancellationToken: stoppingToken);
await foreach (var msg in consumer.ConsumeAsync<byte[]>(cancellationToken: stoppingToken))
{ {
AccountDeletedEvent? evt = null;
try try
{ {
evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data); var evt = JsonSerializer.Deserialize<AccountDeletedEvent>(msg.Data);
if (evt == null) if (evt == null) continue;
{
await msg.AckAsync(cancellationToken: stoppingToken);
continue;
}
logger.LogInformation("Processing account deletion for: {AccountId}", evt.AccountId); logger.LogInformation("Account deleted: {AccountId}", evt.AccountId);
using var scope = serviceProvider.CreateScope(); using var scope = serviceProvider.CreateScope();
var db = scope.ServiceProvider.GetRequiredService<AppDatabase>(); var db = scope.ServiceProvider.GetRequiredService<AppDatabase>();
@@ -142,41 +103,28 @@ public class BroadcastEventHandler(
.Where(p => p.Members.All(m => m.AccountId == evt.AccountId)) .Where(p => p.Members.All(m => m.AccountId == evt.AccountId))
.ToListAsync(cancellationToken: stoppingToken); .ToListAsync(cancellationToken: stoppingToken);
if (publishers.Any()) foreach (var publisher in publishers)
{ await db.Posts
foreach (var publisher in publishers) .Where(p => p.PublisherId == publisher.Id)
{
await db.Posts
.Where(p => p.PublisherId == publisher.Id)
.ExecuteDeleteAsync(cancellationToken: stoppingToken);
}
var publisherIds = publishers.Select(p => p.Id).ToList();
await db.Publishers
.Where(p => publisherIds.Contains(p.Id))
.ExecuteDeleteAsync(cancellationToken: stoppingToken); .ExecuteDeleteAsync(cancellationToken: stoppingToken);
}
var publisherIds = publishers.Select(p => p.Id).ToList();
await db.Publishers
.Where(p => publisherIds.Contains(p.Id))
.ExecuteDeleteAsync(cancellationToken: stoppingToken);
await transaction.CommitAsync(cancellationToken: stoppingToken); await transaction.CommitAsync(cancellationToken: stoppingToken);
await msg.AckAsync(cancellationToken: stoppingToken);
logger.LogInformation("Account deletion for {AccountId} processed successfully in Sphere.",
evt.AccountId);
} }
catch (Exception ex) catch (Exception)
{ {
logger.LogError(ex, await transaction.RollbackAsync(cancellationToken: stoppingToken);
"Error during transaction for account deletion {AccountId} in Sphere, rolling back.",
evt.AccountId);
await transaction.RollbackAsync(CancellationToken.None);
throw; throw;
} }
} }
catch (Exception ex) catch (Exception ex)
{ {
logger.LogError(ex, "Failed to process account deletion for {AccountId} in Sphere, will retry.", logger.LogError(ex, "Error processing AccountDeleted");
evt?.AccountId);
await msg.NakAsync(delay: TimeSpan.FromSeconds(30), cancellationToken: stoppingToken);
} }
} }
} }
} }