⚙️ Publisher-Side Idempotency (بگذارید Broker آن را مدیریت کند)

بسیاری از message brokerها از قبل از طریق قابلیت message deduplication، از انتشار idempotent پشتیبانی می‌کنند؛ البته اگر در پیام خود یک شناسه‌ی یکتا (unique message ID) قرار دهید.

به‌عنوان مثال، Azure Service Bus می‌تواند پیام‌های تکراری را تشخیص داده و انتشار مجدد پیام‌هایی با MessageId یکسان را در بازه‌ی زمانی مشخص‌شده نادیده بگیرد.Amazon SQS و سایر brokerها نیز تضمین‌های مشابهی ارائه می‌دهند. 💪

✅ شما نیازی ندارید این منطق را دوباره در برنامه‌ی خود پیاده‌سازی کنید.
نکته‌ی کلیدی این است که برای هر پیام، یک شناسه‌ی پایدار (stable identifier) اختصاص دهید که به‌طور یکتا رویداد منطقی مورد نظر را نمایش دهد.

به‌عنوان مثال، هنگام انتشار یک event از نوع NoteCreated:

var message = new NoteCreated(note.Id, note.Title, note.Content)
{
    MessageId = Guid.NewGuid() // یا می‌توانید از note.Id استفاده کنید
};

await publisher.PublishAsync(message);

اگر شبکه پس از ارسال پیام قطع شود 🌐، ممکن است برنامه‌ی شما ارسال را مجدداً تلاش کند (retry).
اما زمانی که broker همان MessageId را مشاهده کند، متوجه می‌شود که این پیام تکراری است و آن را به‌صورت ایمن نادیده می‌گیرد. ✅

به این ترتیب، شما deduplication را بدون نیاز به جداول ردیابی (tracking tables) یا state اضافی در سرویس خود به‌دست می‌آورید.

این نوع idempotency در سطح broker، بخش بزرگی از مشکلات سمت producer را حل می‌کند مثل:

• retryهای شبکه 🔁
• خطاهای موقتی ⚠️
• انتشارهای تکراری 🌀

اما چیزی که این مکانیسم پوشش نمی‌دهد، retryهای سمت consumer است یعنی زمانی که پیام‌ها مجدداً تحویل داده می‌شوند یا سرویس شما هنگام پردازش دچار crash می‌شود 💥.

اینجاست که الگوی Idempotent Consumer وارد عمل می‌شود. 🎯

🧠 Implementing an Idempotent Consumer in .NET

در اینجا یک نمونه از Idempotent Consumer برای eventی از نوع NoteCreated آورده شده است:

internal sealed class NoteCreatedConsumer(
    TagsDbContext dbContext,
    HybridCache hybridCache,
    ILogger<Program> logger) : IConsumer<NoteCreated>
{
    public async Task ConsumeAsync(ConsumeContext<NoteCreated> context)
    {
        // 1. بررسی اینکه آیا این پیام قبلاً توسط این consumer پردازش شده است
        if (await dbContext.MessageConsumers.AnyAsync(c =>
                c.MessageId == context.MessageId &&
                c.ConsumerName == nameof(NoteCreatedConsumer)))
        {
            return;
        }

        var request = new AnalyzeNoteRequest(
            context.Message.NoteId,
            context.Message.Title,
            context.Message.Content);

        try
        {
            using var transaction = await dbContext.Database.BeginTransactionAsync();

            // 2. پردازش قطعی (Deterministic): استخراج تگ‌ها از محتوای یادداشت
            var tags = AnalyzeContentForTags(request.Title, request.Content);

            // 3. ذخیره‌سازی تگ‌ها در پایگاه داده
            var tagEntities = tags.Select(ProjectToTagEntity(request.NoteId)).ToList();
            dbContext.Tags.AddRange(tagEntities);

            // 4. ثبت اینکه این پیام پردازش شده است
            dbContext.MessageConsumers.Add(new MessageConsumer
            {
                MessageId = context.MessageId,
                ConsumerName = nameof(NoteCreatedConsumer),
                ConsumedAtUtc = DateTime.UtcNow
            });

            await dbContext.SaveChangesAsync();
            await transaction.CommitAsync();

            // 5. به‌روزرسانی Cache
            await CacheNoteTags(request, tags);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Error analyzing note {NoteId}", request.NoteId);
            throw;
        }
    }
}

این یک نمونه‌ی متداول از Idempotent Consumer است که شامل چند نکته‌ی کلیدی است 🧩