using System.Text.Json; using DysonNetwork.Pass.Wallet; using DysonNetwork.Shared.Stream; using NATS.Client.Core; namespace DysonNetwork.Pass.Startup; public class BroadcastEventHandler( INatsConnection nats, ILogger logger, IServiceProvider serviceProvider ) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await foreach (var msg in nats.SubscribeAsync(PaymentOrderEvent.Type, cancellationToken: stoppingToken)) { PaymentOrderEvent? evt = null; try { evt = JsonSerializer.Deserialize(msg.Data); if (evt?.ProductIdentifier is null || !evt.ProductIdentifier.StartsWith(SubscriptionType.StellarProgram)) { continue; } logger.LogInformation("Handling stellar program order: {OrderId}", evt.OrderId); await using var scope = serviceProvider.CreateAsyncScope(); var db = scope.ServiceProvider.GetRequiredService(); var subscriptions = scope.ServiceProvider.GetRequiredService(); var order = await db.PaymentOrders.FindAsync( [evt.OrderId], cancellationToken: stoppingToken ); if (order is null) { logger.LogWarning("Order with ID {OrderId} not found.", evt.OrderId); continue; } await subscriptions.HandleSubscriptionOrder(order); logger.LogInformation("Subscription for order {OrderId} handled successfully.", evt.OrderId); } catch (Exception ex) { logger.LogError(ex, "Error processing payment order event for order {OrderId}", evt?.OrderId); } } } }