مدیریت رویدادهای از‌دست‌رفته (Handling Missed Events) 🔄📡

ءEndpoint ساده‌ای که همین الان ساختیم عالی است، اما یک ضعف مهم دارد:
تاب‌آوری (Resilience) ندارد.

یکی از بزرگ‌ترین چالش‌ها در استریم‌های Real-time، قطع شدن اتصال است.
تا زمانی که مرورگر به‌صورت خودکار دوباره وصل شود، ممکن است چندین رویداد ارسال شده و از دست رفته باشند 😕

برای حل این مشکل، SSE یک مکانیزم داخلی دارد:
هدر Last-Event-ID.
وقتی مرورگر reconnect می‌شود، این ID را دوباره برای سرور ارسال می‌کند.

در NET 10. می‌توانیم از نوع <SseItem<T استفاده کنیم تا داده را به‌همراه متادیتاهایی مثل ID و retry interval بسته‌بندی کنیم.

با ترکیب یک OrderEventBuffer ساده‌ی درون‌حافظه‌ای و مقدار Last-Event-ID که مرورگر ارسال می‌کند، می‌توانیم رویدادهای از‌دست‌رفته را هنگام reconnect دوباره ارسال کنیم 🔁

app.MapGet("orders/realtime/with-replays", (
  ChannelReader<OrderPlacement> channelReader,
  OrderEventBuffer eventBuffer,
  [FromHeader(Name = "Last-Event-ID")] string? lastEventId,
  CancellationToken cancellationToken) =>
{
  async IAsyncEnumerable<SseItem<OrderPlacement>> StreamEvents()
  {
    // 1. بازپخش رویدادهای از‌دست‌رفته از buffer
    if (!string.IsNullOrWhiteSpace(lastEventId))
    {
      var missedEvents = eventBuffer.GetEventsAfter(lastEventId);
      foreach (var missedEvent in missedEvents)
      {
        yield return missedEvent;
      }
    }

    // 2. استریم رویدادهای جدید به‌محض ورود به Channel
    await foreach (var order in channelReader.ReadAllAsync(cancellationToken))
    {
      var sseItem = eventBuffer.Add(order); // Buffer یک ID یکتا اختصاص می‌دهد
      yield return sseItem;
    }
  }

  return TypedResults.ServerSentEvents(StreamEvents(), "orders");
});

فیلتر کردن Server-Sent Events بر اساس کاربر 👤🔐

ءSSE روی HTTP استاندارد ساخته شده است.
چون یک درخواست GET معمولی است، زیرساخت فعلی شما بدون تغییر کار می‌کند:

ءSecurity 🔐: می‌توانید JWT را به‌صورت عادی در هدر Authorization ارسال کنید

ءUser Context 👤: می‌توانید به HttpContext.User دسترسی داشته باشید و استریم را بر اساس UserId فیلتر کنید
→ فقط داده‌هایی که متعلق به همان کاربر هستند ارسال می‌شوند

مثال یک Endpoint SSE که فقط سفارش‌های کاربر لاگین‌شده را استریم می‌کند:

app.MapGet("orders/realtime", (
  ChannelReader<OrderPlacement> channelReader,
  IUserContext userContext, // کانتکست تزریق‌شده شامل اطلاعات کاربر
  CancellationToken cancellationToken) =>
{
  // UserId از JWT توسط IUserContext استخراج می‌شود
  var currentUserId = userContext.UserId;

  async IAsyncEnumerable<OrderPlacement> GetUserOrders()
  {
    await foreach (var order in channelReader.ReadAllAsync(cancellationToken))
    {
      // فقط داده‌های متعلق به کاربر احراز هویت‌شده ارسال می‌شود
      if (order.CustomerId == currentUserId)
      {
        yield return order;
      }
    }
  }

  return Results.ServerSentEvents(GetUserOrders(), "orders");
})
.RequireAuthorization(); // Authorization استاندارد ASP.NET Core

⚠️ نکته مهم:

وقتی یک پیام داخل Channel نوشته می‌شود، به تمام کلاینت‌های متصل broadcast می‌شود.
این رفتار برای استریم‌های per-user ایده‌آل نیست.
در محیط production، احتمالاً به راهکار قوی‌تری نیاز دارید.

مصرف Server-Sent Events در JavaScript 🌐🧠

در سمت کلاینت، نیازی به نصب حتی یک پکیج npm هم ندارید 🙌 API بومی مرورگر یعنی EventSource تمام کارهای سنگین را انجام می‌دهد،
از جمله reconnect خودکار و ارسال Last-Event-ID.

const eventSource = new EventSource('/orders/realtime/with-replays');

// گوش دادن به event type مشخص‌شده در C#
eventSource.addEventListener('orders', (event) => {
  const payload = JSON.parse(event.data);
  console.log(New Order ${event.lastEventId}:, payload.data);
});

// وقتی اتصال برقرار می‌شود
eventSource.onopen = () => {
  console.log('Connection opened');
};

// پیام‌های عمومی (در صورت وجود)
eventSource.onmessage = (event) => {
  console.log('Received message:', event);
};

// مدیریت خطا و reconnect
eventSource.onerror = () => {
  if (eventSource.readyState === EventSource.CONNECTING) {
    console.log('Reconnecting...');
  }
};