Message Reliability

// Channel'da confirms aktifleştir
await channel.ConfirmSelectAsync();

// Publish
await channel.BasicPublishAsync(
    exchange: "orders",
    routingKey: "order.created",
    mandatory: true,
    basicProperties: new BasicProperties
    {
        DeliveryMode = DeliveryModes.Persistent,
        MessageId = Guid.NewGuid().ToString(),
        Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
    },
    body: messageBody);

// Confirm bekle (batch)
await channel.WaitForConfirmsOrDieAsync(TimeSpan.FromSeconds(5));
// Eğer confirm gelmezse exception fırlatır → retry logic tetikle

// Mesaj: Persistent (diske yazılsın)
var props = new BasicProperties
{
    DeliveryMode = DeliveryModes.Persistent,  // ← BUNU UNUTMA
    MessageId = Guid.NewGuid().ToString()
};

// Queue: Durable (restart'ta kaybolmasın)
await channel.QueueDeclareAsync(
    queue: "orders",
    durable: true,       // ← BUNU UNUTMA
    exclusive: false,
    autoDelete: false);

var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (sender, ea) =>
{
    try
    {
        var body = ea.Body.ToArray();
        var order = JsonSerializer.Deserialize<OrderEvent>(body);

        // İş mantığı — idempotent olmalı
        await _orderProcessor.ProcessAsync(order!);

        // Başarılı → Ack
        await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
    }
    catch (TransientException)
    {
        // Geçici hata → requeue (tekrar denenecek)
        await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: true);
    }
    catch (PermanentException)
    {
        // Kalıcı hata → DLX'e gönder (requeue: false)
        await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
    }
};

await channel.BasicConsumeAsync(
    queue: "payment-processing",
    autoAck: false,  // ← MANUAL ACK
    consumer: consumer);

public class IdempotentOrderConsumer
{
    private readonly AppDbContext _db;

    public async Task HandleAsync(IChannel channel, BasicDeliverEventArgs ea)
    {
        var messageId = ea.BasicProperties.MessageId;

        // 1. Daha önce işlenmiş mi kontrol et
        if (await _db.ProcessedMessages.AnyAsync(m => m.MessageId == messageId))
        {
            // Duplicate — sessizce ack'la, tekrar işleme
            await channel.BasicAckAsync(ea.DeliveryTag, false);
            return;
        }

        try
        {
            var order = JsonSerializer.Deserialize<OrderEvent>(ea.Body.Span);

            // 2. İş mantığı + idempotency kaydı AYNI TRANSACTION'da
            await using var tx = await _db.Database.BeginTransactionAsync();

            await _orderService.ProcessAsync(order!);

            _db.ProcessedMessages.Add(new ProcessedMessage
            {
                MessageId = messageId,
                ProcessedAt = DateTime.UtcNow
            });
            await _db.SaveChangesAsync();
            await tx.CommitAsync();

            // 3. Başarılıysa ack
            await channel.BasicAckAsync(ea.DeliveryTag, false);
        }
        catch (Exception)
        {
            await channel.BasicNackAsync(ea.DeliveryTag, false, requeue: true);
        }
    }
}

// Entity
public class ProcessedMessage
{
    public string MessageId { get; set; } = "";
    public DateTime ProcessedAt { get; set; }
}
// Migration: CREATE UNIQUE INDEX IX_ProcessedMessages_MessageId ON ProcessedMessages(MessageId);