Integration & messaging

// Kafka consumer: each consumer group gets every event and tracks its own offset
var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "fraud-service",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false            // commit only after we've handled the event
};
using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("payments");

while (!ct.IsCancellationRequested)
{
    var result = consumer.Consume(ct);  // blocks until a record arrives
    await HandleAsync(result.Message.Key, result.Message.Value, ct);
    consumer.Commit(result);            // move this group's offset forward
}

var processor = client.CreateProcessor("payments", new ServiceBusProcessorOptions
{
    AutoCompleteMessages = false,   // peek-lock: we settle each message ourselves
    MaxConcurrentCalls = 8
});
processor.ProcessMessageAsync += async args =>
{
    var msg = args.Message;
    var payment = msg.Body.ToObjectFromJson<PaymentDto>();
    if (payment is null)            // poison message: never retry it
    {
        await args.DeadLetterMessageAsync(msg, "InvalidBody", "Could not parse payment");
        return;
    }
    await using var scope = scopes.CreateAsyncScope();
    var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
    if (!await db.ProcessedMessages.AnyAsync(p => p.Id == msg.MessageId))
    {
        db.ProcessedMessages.Add(new ProcessedMessage(msg.MessageId)); // unique key
        db.Payments.Add(Payment.From(payment));
        await db.SaveChangesAsync(args.CancellationToken);  // one transaction
    }
    await args.CompleteMessageAsync(msg);  // ack only after the DB commit
};
processor.ProcessErrorAsync += args => Task.CompletedTask; // log args.Exception
await processor.StartProcessingAsync();

// Bounded channel: when 100 items are waiting, the producer has to wait
var channel = Channel.CreateBounded<Payment>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait,   // back-pressure, nothing is dropped
    SingleWriter = true
});

// Producer: WriteAsync only completes when there is room
async Task ProduceAsync(IAsyncEnumerable<Payment> incoming, CancellationToken ct)
{
    await foreach (var p in incoming.WithCancellation(ct))
        await channel.Writer.WriteAsync(p, ct);  // slows to the consumers' pace
    channel.Writer.Complete();
}

// Consumers: 4 workers drain the channel at their own pace
var workers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (var p in channel.Reader.ReadAllAsync(ct))
        await ProcessAsync(p, ct);
}));
await Task.WhenAll(workers);

A channel lives in memory, so it's lost on restart. Use it to protect one process. Business messages that must survive belong in a durable broker (Service Bus, Kafka).

// 1. One SaveChanges = one transaction: the order AND its outbox row
db.Orders.Add(order);
db.Outbox.Add(new OutboxMessage { Id = Guid.NewGuid(), Type = "OrderCreated",
    Payload = JsonSerializer.Serialize(new OrderCreated(order.Id)) });
await db.SaveChangesAsync(ct);

// 2. Relay: publish unsent rows to the broker, then mark them sent
public sealed class OutboxRelay(IServiceScopeFactory scopes, ServiceBusSender sender)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(2));
        while (await timer.WaitForNextTickAsync(ct))
        {
            await using var scope = scopes.CreateAsyncScope();
            var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
            var batch = await db.Outbox.Where(m => m.SentAt == null)
                .OrderBy(m => m.CreatedAt).Take(100).ToListAsync(ct);
            foreach (var m in batch)
            {
                await sender.SendMessageAsync(new ServiceBusMessage(m.Payload)
                    { MessageId = m.Id.ToString(), Subject = m.Type }, ct);
                m.SentAt = DateTime.UtcNow;
            }
            await db.SaveChangesAsync(ct);
        }
    }
}

The relay can crash after sending but before saving SentAt, so a message may go out twice. Reuse the outbox id as the MessageId and keep consumers idempotent.

public static class OrderWorkflow
{
    [Function(nameof(RunOrder))]
    public static async Task<string> RunOrder(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        var order = context.GetInput<Order>()!;
        // The orchestrator calls each step in order. State is checkpointed
        // after every await, so it survives restarts.
        await context.CallActivityAsync("TakePayment", order);
        await context.CallActivityAsync("ReserveStock", order);
        return await context.CallActivityAsync<string>("BookShipping", order);
    }

    [Function("TakePayment")]
    public static Task TakePayment([ActivityTrigger] Order order) => Task.CompletedTask;
}

// Start it from an HTTP function or API:
// await client.ScheduleNewOrchestrationInstanceAsync(nameof(OrderWorkflow.RunOrder), order);

Orchestrator code must be deterministic: no DateTime.UtcNow, Guid.NewGuid() or direct I/O. Use context.CurrentUtcDateTime and do I/O in activities.

using var sftp = new SftpClient(host, "svc-batch", new PrivateKeyFile(keyPath));
sftp.Connect();

foreach (var file in sftp.ListDirectory("/inbound").Where(f => f.Name.EndsWith(".csv")))
{
    // Only pick up files the sender has finished writing
    if (!sftp.Exists(file.FullName + ".done")) continue;

    var local = Path.Combine(workDir, file.Name);
    using (var output = File.Create(local))
        sftp.DownloadFile(file.FullName, output);

    await ProcessWithCheckpointsAsync(local, ct);           // validate, load, reconcile
    sftp.RenameFile(file.FullName, $"/archive/{file.Name}"); // never process it twice
}

using var producer = new ProducerBuilder<string, string>(new ProducerConfig
{
    BootstrapServers = servers,
    Acks = Acks.All,
    EnableIdempotence = true          // librdkafka default is false: set it explicitly
}).Build();

await producer.ProduceAsync("payments",
    new Message<string, string> { Key = accountId, Value = json });   // key = ordering scope

using var consumer = new ConsumerBuilder<string, string>(new ConsumerConfig
{
    BootstrapServers = servers,
    GroupId = "ledger-service",
    EnableAutoCommit = false,         // process, then commit = at-least-once
    PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky
}).Build();
consumer.Subscribe("payments");

while (!ct.IsCancellationRequested)
{
    var r = consumer.Consume(ct);
    try { await HandleIdempotentlyAsync(r.Message, ct); }
    catch (Exception ex) when (ex is not OperationCanceledException)
    {
        await producer.ProduceAsync("payments.DLT", new Message<string, string>
        {
            Key = r.Message.Key, Value = r.Message.Value,
            Headers = new Headers { { "error", Encoding.UTF8.GetBytes(ex.Message) } }
        }, ct);                       // park it, don't block the partition
    }
    consumer.Commit(r);
}