Azure Service Bus — Reliable Async Messaging

In-process MediatR events (modular monolith):
  → Same process, in-memory — no network
  → No durability: if the process crashes, the event is lost
  → No cross-service delivery: only modules in the same process can subscribe

Azure Service Bus (microservices or cross-process):
  → Messages persist until consumed — survives process restarts
  → Multiple consumers (subscriptions) can receive the same message
  → Dead-letter queue for unprocessable messages
  → Order guarantees with sessions

Use Azure Service Bus when:
  → Two services need to communicate reliably
  → The publisher and consumer are in different deployments
  → Messages must not be lost on publisher failure
  → You need to fan out one event to multiple subscribers

// NuGet: Azure.Messaging.ServiceBus

builder.Services.AddSingleton(sp =>
    new ServiceBusClient(
        builder.Configuration["ServiceBus:ConnectionString"],
        new ServiceBusClientOptions
        {
            TransportType = ServiceBusTransportType.AmqpWebSockets
        }));

// Or with Managed Identity (no connection string key):
builder.Services.AddSingleton(sp =>
    new ServiceBusClient(
        fullyQualifiedNamespace: "clinical-servicebus.servicebus.windows.net",
        credential: new DefaultAzureCredential()));

public interface IIntegrationEventBus
{
    Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class;
}

public sealed class ServiceBusIntegrationEventBus : IIntegrationEventBus
{
    private readonly ServiceBusClient _client;

    public ServiceBusIntegrationEventBus(ServiceBusClient client) => _client = client;

    public async Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class
    {
        var topicName = GetTopicName(typeof(T)); // e.g. "prescription-created"
        await using var sender = _client.CreateSender(topicName);

        var message = new ServiceBusMessage(JsonSerializer.SerializeToUtf8Bytes(@event))
        {
            ContentType    = "application/json",
            Subject        = typeof(T).Name,
            MessageId      = Guid.NewGuid().ToString(),
            CorrelationId  = Activity.Current?.TraceId.ToString()
        };

        await sender.SendMessageAsync(message, ct);
    }

    private static string GetTopicName(Type eventType) =>
        eventType.Name
            .Replace("Event", string.Empty)
            .ToLowerInvariant()
            .Replace("integrationevent", string.Empty)
            + "-events"; // "PrescriptionCreatedIntegrationEvent" → "prescriptioncreated-events"
}

// Example: publish prescription created event after saving to DB
await _eventBus.PublishAsync(new PrescriptionCreatedIntegrationEvent(
    PrescriptionId: prescription.Id.Value,
    PatientId:      prescription.PatientId.Value,
    Medication:     prescription.MedicationName.Value,
    CreatedAt:      DateTime.UtcNow), ct);

public sealed class PrescriptionCreatedConsumer : BackgroundService
{
    private readonly ServiceBusClient _client;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<PrescriptionCreatedConsumer> _logger;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var processor = _client.CreateProcessor(
            topicName:        "prescriptioncreated-events",
            subscriptionName: "pharmacy-service",
            new ServiceBusProcessorOptions
            {
                MaxConcurrentCalls   = 4,
                AutoCompleteMessages = false // manually complete to control acknowledgement
            });

        processor.ProcessMessageAsync += HandleMessageAsync;
        processor.ProcessErrorAsync   += HandleErrorAsync;

        await processor.StartProcessingAsync(stoppingToken);
        await Task.Delay(Timeout.Infinite, stoppingToken);
        await processor.StopProcessingAsync();
    }

    private async Task HandleMessageAsync(ProcessMessageEventArgs args)
    {
        using var scope   = _scopeFactory.CreateScope();
        var handler       = scope.ServiceProvider
            .GetRequiredService<IIntegrationEventHandler<PrescriptionCreatedIntegrationEvent>>();

        var @event = JsonSerializer.Deserialize<PrescriptionCreatedIntegrationEvent>(
            args.Message.Body)!;

        await handler.HandleAsync(@event, args.CancellationToken);

        // Complete the message — removes it from the queue
        await args.CompleteMessageAsync(args.Message);
    }

    private Task HandleErrorAsync(ProcessErrorEventArgs args)
    {
        _logger.LogError(args.Exception,
            "Service Bus error on {Source}", args.ErrorSource);
        return Task.CompletedTask;
        // Message is NOT completed — Service Bus retries based on MaxDeliveryCount
    }
}

// Messages that fail MaxDeliveryCount times are moved to the dead-letter queue (DLQ)
// Default MaxDeliveryCount: 10

// Process dead-lettered messages for investigation and manual reprocessing:
var dlqProcessor = _client.CreateProcessor(
    topicName:        "prescriptioncreated-events",
    subscriptionName: "$DeadLetterQueue/pharmacy-service");

// Monitor DLQ message count — if it grows, something is systematically failing
// Alert: "DLQ count > 0 for more than 5 minutes" in Azure Monitor

// Inspect DLQ messages:
await using var receiver = _client.CreateReceiver(
    "prescriptioncreated-events",
    "$DeadLetterQueue/pharmacy-service");

var dlqMessages = await receiver.ReceiveMessagesAsync(maxMessages: 20);
foreach (var msg in dlqMessages)
{
    _logger.LogError(
        "DLQ message {MessageId}: DeadLetterReason={Reason}, Body={Body}",
        msg.MessageId,
        msg.DeadLetterReason,
        msg.Body.ToString());
}

// Use sessions when messages for the same patient must be processed in order
// All messages with the same SessionId are processed by the same consumer instance

var message = new ServiceBusMessage(payload)
{
    SessionId = patientId.ToString() // all messages for this patient go to same session
};

// Session processor guarantees order within a session
var sessionProcessor = _client.CreateSessionProcessor(
    topicName:        "patient-events",
    subscriptionName: "audit-service",
    new ServiceBusSessionProcessorOptions
    {
        MaxConcurrentSessions = 10 // process up to 10 patients' events concurrently
    });