// ---------------------------------------------------------------------------- // BroadcastNotificationEndpoint // // Demonstrates: // - IRawStellaEndpoint for streaming response // - Bulk notification delivery with progress // - NDJSON streaming format // - Long-running operation with heartbeat // ---------------------------------------------------------------------------- using Examples.NotificationService.Models; using Microsoft.Extensions.Logging; using StellaOps.Microservice; using System.Text; using System.Text.Json; namespace Examples.NotificationService.Endpoints; /// /// Streaming endpoint for broadcasting notifications to multiple recipients. /// Returns progress updates as NDJSON stream. /// [StellaEndpoint("POST", "/notifications/broadcast", SupportsStreaming = true, TimeoutSeconds = 600)] public sealed class BroadcastNotificationEndpoint : IRawStellaEndpoint { private readonly ILogger _logger; public BroadcastNotificationEndpoint(ILogger logger) { _logger = logger; } public async Task HandleAsync( RawRequestContext context, CancellationToken cancellationToken) { // Parse the broadcast request from body using var reader = new StreamReader(context.Body, Encoding.UTF8); var requestJson = await reader.ReadToEndAsync(cancellationToken); var request = JsonSerializer.Deserialize(requestJson, new JsonSerializerOptions { PropertyNameCaseInsensitive = true }); if (request == null || string.IsNullOrEmpty(request.Title)) { return RawResponse.BadRequest("Invalid broadcast request"); } _logger.LogInformation( "Starting broadcast. Type: {Type}, Recipients: {Count}, CorrelationId: {CorrelationId}", request.Type, request.RecipientIds?.Length ?? 0, context.CorrelationId); // Create response stream for NDJSON progress updates var stream = new MemoryStream(); var writer = new StreamWriter(stream, Encoding.UTF8, leaveOpen: true); var broadcastId = $"BCAST-{DateTime.UtcNow:yyyyMMdd}-{Guid.NewGuid():N}"[..26].ToUpperInvariant(); var recipients = request.RecipientIds ?? GenerateSimulatedRecipients(request.RecipientCount ?? 100); var totalRecipients = recipients.Length; var deliveredCount = 0; var failedCount = 0; try { // Write initial progress await WriteProgressAsync(writer, new BroadcastProgress { BroadcastId = broadcastId, Status = "started", TotalRecipients = totalRecipients, DeliveredCount = 0, FailedCount = 0, PercentComplete = 0, Timestamp = DateTimeOffset.UtcNow }, cancellationToken); // Process recipients in batches const int batchSize = 10; var random = new Random(); for (var i = 0; i < recipients.Length; i += batchSize) { cancellationToken.ThrowIfCancellationRequested(); var batch = recipients.Skip(i).Take(batchSize).ToArray(); // Simulate sending to each recipient in batch foreach (var recipientId in batch) { // Simulate occasional failures (5%) if (random.NextDouble() < 0.05) { failedCount++; _logger.LogDebug("Failed to deliver to {RecipientId}", recipientId); } else { deliveredCount++; } } // Simulate processing time await Task.Delay(TimeSpan.FromMilliseconds(50 + random.Next(100)), cancellationToken); // Write progress update var percentComplete = (int)((i + batch.Length) * 100.0 / totalRecipients); await WriteProgressAsync(writer, new BroadcastProgress { BroadcastId = broadcastId, Status = "in_progress", TotalRecipients = totalRecipients, DeliveredCount = deliveredCount, FailedCount = failedCount, PercentComplete = percentComplete, CurrentBatch = i / batchSize + 1, TotalBatches = (totalRecipients + batchSize - 1) / batchSize, Timestamp = DateTimeOffset.UtcNow }, cancellationToken); } // Write completion await WriteProgressAsync(writer, new BroadcastProgress { BroadcastId = broadcastId, Status = "completed", TotalRecipients = totalRecipients, DeliveredCount = deliveredCount, FailedCount = failedCount, PercentComplete = 100, Timestamp = DateTimeOffset.UtcNow, Summary = new BroadcastSummary { SuccessRate = (decimal)deliveredCount / totalRecipients * 100, Duration = TimeSpan.FromSeconds(totalRecipients * 0.015) // Simulated } }, cancellationToken); _logger.LogInformation( "Broadcast {BroadcastId} completed. Delivered: {Delivered}, Failed: {Failed}", broadcastId, deliveredCount, failedCount); } catch (OperationCanceledException) { await WriteProgressAsync(writer, new BroadcastProgress { BroadcastId = broadcastId, Status = "cancelled", TotalRecipients = totalRecipients, DeliveredCount = deliveredCount, FailedCount = failedCount, PercentComplete = (int)(deliveredCount * 100.0 / totalRecipients), Timestamp = DateTimeOffset.UtcNow }, CancellationToken.None); _logger.LogWarning("Broadcast {BroadcastId} cancelled", broadcastId); } await writer.FlushAsync(CancellationToken.None); stream.Position = 0; var headers = new HeaderCollection(); headers.Set("Content-Type", "application/x-ndjson"); headers.Set("X-Broadcast-Id", broadcastId); return new RawResponse { StatusCode = 200, Headers = headers, Body = stream }; } private static async Task WriteProgressAsync(StreamWriter writer, BroadcastProgress progress, CancellationToken cancellationToken) { var json = JsonSerializer.Serialize(progress, new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }); await writer.WriteLineAsync(json); await writer.FlushAsync(cancellationToken); } private static string[] GenerateSimulatedRecipients(int count) { return Enumerable.Range(1, count) .Select(i => $"USR-{i:D6}") .ToArray(); } private sealed record BroadcastRequest { public required string Type { get; init; } public required string Title { get; init; } public required string Body { get; init; } public NotificationPriority Priority { get; init; } = NotificationPriority.Normal; public DeliveryChannel[] Channels { get; init; } = [DeliveryChannel.InApp]; public string[]? RecipientIds { get; init; } public int? RecipientCount { get; init; } } private sealed record BroadcastProgress { public required string BroadcastId { get; init; } public required string Status { get; init; } public int TotalRecipients { get; init; } public int DeliveredCount { get; init; } public int FailedCount { get; init; } public int PercentComplete { get; init; } public int? CurrentBatch { get; init; } public int? TotalBatches { get; init; } public required DateTimeOffset Timestamp { get; init; } public BroadcastSummary? Summary { get; init; } } private sealed record BroadcastSummary { public decimal SuccessRate { get; init; } public TimeSpan Duration { get; init; } } }