Перейти к основному содержимому

SignalR Realtime — Архитектура и Практическое Руководство

Назначение​

Документ описывает архитектуру realtime push-уведомлений в Pin через SignalR в условиях multi-replica Kubernetes deployment. Описывает выбор backplane, пакеты, интерфейсы и паттерны публикации уведомлений из background workers и других микросервисов.

Проблема: Multi-Replica Kubernetes​

В k8s у каждого сервиса может быть несколько pod-ов (replicas). SignalR Connection хранится только на одном pod-е, к которому подключён браузер игрока. Если уведомление о пополнении баланса приходит в другой pod — оно потеряется.

Browser ──────WebSocket──── pod-1 (has connection)
                             pod-2 (NO connection — message lost without backplane)
                             pod-3 (NO connection)

Решение: Redis Backplane​

Browser ──WebSocket── pod-1 ──publish──► Redis PubSub
                      pod-2 ──subscribe──► Redis PubSub ──► broadcast к своим connections
                      pod-3 ──subscribe──► Redis PubSub

AddSignalR().AddStackExchangeRedis(connectionString) — все pod-ы подписываются на Redis channel; при публикации сообщение доставляется всем.

Пакеты​

src/building-blocks/
  Talent.Common.Realtime.Core/     ← IRealtimeNotificationPublisher, message contracts
  Talent.Common.Realtime.Hub/      ← PinHubBase<T>, backplane setup, DI extensions
  Talent.Common.Realtime.Client/   ← IRealtimeProducer для worker/other-service → MQ

Talent.Common.Realtime.Core​

NuGet dependencies: только BuildingBlocks.Application (нет ASP.NET, нет Redis прямых зависимостей).

Talent.Common.Realtime.Core/
  Contracts/
    IRealtimeNotificationPublisher.cs   — отправить уведомление игроку/партнёру по ID
    IRealtimeGroupPublisher.cs          — отправить в группу (tenant, brand)
    RealtimeMessage.cs                  — typed message contract
    RealtimeChannel.cs                  — константы channel names
  Messages/
    BalanceUpdatedMessage.cs
    WithdrawalStatusChangedMessage.cs
    IncentiveActivatedMessage.cs
    CompetitionLeaderboardUpdatedMessage.cs
    SystemNotificationMessage.cs

Talent.Common.Realtime.Hub​

NuGet dependencies: Microsoft.AspNetCore.SignalR, Microsoft.AspNetCore.SignalR.StackExchangeRedis, Realtime.Core.

Talent.Common.Realtime.Hub/
  PinHubBase.cs                    — base Hub с auth, tenant scope, rate limit
  SignalRNotificationPublisher.cs      — реализация IRealtimeNotificationPublisher через IHubContext
  BackplaneOptions.cs
  DependencyInjection/
    AddPinSignalR.cs               — регистрация + Redis backplane
    AddPinHubs.cs

Talent.Common.Realtime.Client​

NuGet dependencies: MassTransit (или только StackExchange.Redis), Realtime.Core.
Используется в background workers и других сервисах, у которых нет SignalR Hub.

Talent.Common.Realtime.Client/
  MqRealtimeProducer.cs               — публикует RealtimeEnvelope в Redis PubSub / RabbitMQ
  IRealtimeProducer.cs
  DependencyInjection/
    AddPinRealtimeClient.cs

Интерфейсы​

IRealtimeNotificationPublisher​

// Talent.Common.Realtime.Core/Contracts/IRealtimeNotificationPublisher.cs

/// <summary>
/// Публикует realtime push-сообщение конкретному пользователю или группе.
/// Используется только в сервисах, у которых есть SignalR Hub (т.е. в public-api / user-api).
/// Для других сервисов и background workers — используй <see cref="IRealtimeProducer"/>.
/// </summary>
public interface IRealtimeNotificationPublisher
{
    /// <summary>Отправить конкретному игроку по UserId (на все его devices).</summary>
    Task SendToUser<TMessage>(
        UserId userId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class;

    /// <summary>Отправить всем подключённым игрокам данного бренда.</summary>
    Task SendToBrand<TMessage>(
        BrandId brandId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class;

    /// <summary>Отправить конкретному admin/partner пользователю.</summary>
    Task SendToUser<TMessage>(
        string userId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class;
}

IRealtimeProducer (для worker / другой сервис → MQ → Hub)​

// Talent.Common.Realtime.Client/IRealtimeProducer.cs

/// <summary>
/// Отправляет realtime-уведомление через MQ (Redis PubSub или RabbitMQ).
/// Используется в background workers и микросервисах без SignalR Hub.
/// User-api / public-api подписывается на MQ и доставляет через Hub.
/// </summary>
public interface IRealtimeProducer
{
    Task NotifyUser<TMessage>(
        UserId userId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class;

    Task NotifyBrand<TMessage>(
        BrandId brandId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class;
}

PinHubBase — Базовый Hub​

// Talent.Common.Realtime.Hub/PinHubBase.cs

/// <summary>
/// Базовый SignalR Hub. Обеспечивает:
/// - JWT-аутентификацию и извлечение UserId/BrandId из claims.
/// - Автоматическое добавление в группы при Connect (user-{id}, brand-{id}, tenant-{id}).
/// - Rate limiting подключений.
/// - Structured logging connect/disconnect.
/// </summary>
public abstract class PinHubBase<T> : Hub<T>
    where T : class
{
    private readonly IIdentityContextProvider _identityProvider;
    private readonly ILogger _logger;

    public override async Task OnConnectedAsync()
    {
        var identity = _identityProvider.TryGetCurrent();
        if (identity is null)
        {
            Context.Abort();
            return;
        }

        // Добавляем в typed groups для таргетированной отправки
        if (identity.ActorType == ActorType.User && identity.ActorId is not null)
            await Groups.AddToGroupAsync(Context.ConnectionId, $"user:{identity.ActorId}");

        if (identity.BrandId is not null)
            await Groups.AddToGroupAsync(Context.ConnectionId, $"brand:{identity.BrandId}");

        if (identity.TenantId is not null)
            await Groups.AddToGroupAsync(Context.ConnectionId, $"tenant:{identity.TenantId}");

        _logger.LogInformation(
            "SignalR connected actorType={ActorType} actorId={ActorId} connectionId={ConnectionId}",
            identity.ActorType, identity.ActorId, Context.ConnectionId);

        await base.OnConnectedAsync();
    }

    public override async Task OnDisconnectedAsync(Exception? exception)
    {
        _logger.LogInformation(
            "SignalR disconnected connectionId={ConnectionId} error={Error}",
            Context.ConnectionId, exception?.Message);

        await base.OnDisconnectedAsync(exception);
    }
}

// Конкретный Hub в user-api:
public sealed class UserNotificationHub : PinHubBase<IUserNotificationClient>
{
    public UserNotificationHub(
        IIdentityContextProvider identityProvider,
        ILogger<UserNotificationHub> logger)
        : base(identityProvider, logger) { }
}

// Typed client interface (методы, которые браузер может получить):
public interface IUserNotificationClient
{
    Task BalanceUpdated(BalanceUpdatedMessage message);
    Task WithdrawalStatusChanged(WithdrawalStatusChangedMessage message);
    Task IncentiveActivated(IncentiveActivatedMessage message);
    Task SystemNotification(SystemNotificationMessage message);
}

Регистрация в DI​

// user-api / Pin.User.Api/Startup/UserWebStartup.cs

protected override void ConfigureSignalR(IPinApiBuilder builder)
{
    builder.AddPinSignalR(options =>
    {
        // Redis backplane — обязателен в k8s multi-replica
        options.UseRedisBackplane(
            connectionString: Configuration.GetConnectionString("Redis")!);

        // Настройки Hub
        options.MaximumReceiveMessageSize = 32 * 1024; // 32 KB
        options.ClientTimeoutInterval = TimeSpan.FromSeconds(30);
        options.KeepAliveInterval = TimeSpan.FromSeconds(15);
    });

    builder.AddHub<UserNotificationHub>("/hubs/user");
}

Typed Message Contracts​

// Talent.Common.Realtime.Core/Messages/

public sealed record BalanceUpdatedMessage
{
    public required string AccountingAccountId { get; init; }
    public required decimal NewBalance { get; init; }
    public required short CurrencyId { get; init; }      // ссылка на справочник handbook
    public required string CurrencyCode { get; init; }   // display only (из CurrencySnapshot)
    public required string ChangeReason { get; init; }   // "deposit", "withdrawal", "incentive", "bet"
    public required DateTimeOffset UpdatedAt { get; init; }
}

public sealed record WithdrawalStatusChangedMessage
{
    public required string WithdrawalId { get; init; }
    public required string NewStatus { get; init; }
    public required string? ReasonCode { get; init; }
    public required DateTimeOffset ChangedAt { get; init; }
}

public sealed record IncentiveActivatedMessage
{
    public required string UserIncentiveId { get; init; }
    public required string IncentiveName { get; init; }
    public required decimal IncentiveAmount { get; init; }
    public required short CurrencyId { get; init; }      // ссылка на справочник handbook
    public required decimal WagerRequired { get; init; }
    public required DateTimeOffset ExpiresAt { get; init; }
}

public sealed record SystemNotificationMessage
{
    public required string NotificationId { get; init; }
    public required string Category { get; init; }     // "info", "warning", "promo"
    public required string TitleKey { get; init; }     // i18n key
    public required string BodyKey { get; init; }
    public Dictionary<string, string> Params { get; init; } = [];
    public DateTimeOffset? ExpiresAt { get; init; }
}

// Константы channel names — избегаем magic strings
public static class RealtimeEvents
{
    public const string BalanceUpdated             = "balance_updated";
    public const string WithdrawalStatusChanged    = "withdrawal_status_changed";
    public const string IncentiveActivated             = "incentive_activated";
    public const string IncentiveExpiringSoon          = "incentive_expiring_soon";
    public const string CompetitionLeaderboard      = "competition_leaderboard_updated";
    public const string SystemNotification         = "system_notification";
    public const string GameSessionExpired         = "game_session_expired";
}

Поток: Background Worker → SignalR Push​

Пример: accounting-worker завершает депозит и уведомляет игрока через user-api.

accounting-worker
  → DepositCallbackConsumer.Handle()
  → _depositService.CompleteDeposit(...)        // обновляет balance
  → _realtimeProducer.NotifyUser(             // публикует в Redis PubSub
        userId,
        RealtimeEvents.BalanceUpdated,
        new BalanceUpdatedMessage { ... })

Redis PubSub channel: "Pin:realtime:user:{userId}"
  ↓ (все pod-ы user-api подписаны)

user-api pod-1 (has connection for userId)
  → RealtimeNotificationRelay (IHostedService)
  → _hubContext.Clients.Group($"user:{userId}")
       .SendAsync(RealtimeEvents.BalanceUpdated, message)

Browser WebSocket → BalanceUpdated event

MqRealtimeProducer Implementation​

// Talent.Common.Realtime.Client/MqRealtimeProducer.cs

public sealed class MqRealtimeProducer : IRealtimeProducer
{
    private readonly IConnectionMultiplexer _redis;
    private readonly ILogger<MqRealtimeProducer> _logger;

    public async Task NotifyUser<TMessage>(
        UserId userId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class
    {
        var channel  = $"Pin:realtime:user:{userId.Value}";
        var envelope = JsonSerializer.Serialize(new RealtimeEnvelope
        {
            Channel   = channel,
            EventName = eventName,
            Target    = RealtimeTarget.User,
            TargetId  = userId.Value.ToString(),
            Payload   = JsonSerializer.SerializeToElement(message),
        });

        await _redis.GetSubscriber()
            .PublishAsync(RedisChannel.Literal(channel), envelope);

        _logger.LogDebug(
            "RealtimeProducer published {EventName} to user {UserId}",
            eventName, userId.Value);
    }

    public async Task NotifyBrand<TMessage>(
        BrandId brandId,
        string eventName,
        TMessage message,
        CancellationToken ct = default)
        where TMessage : class
    {
        var channel  = $"Pin:realtime:brand:{brandId.Value}";
        var envelope = JsonSerializer.Serialize(new RealtimeEnvelope
        {
            Channel   = channel,
            EventName = eventName,
            Target    = RealtimeTarget.Brand,
            TargetId  = brandId.Value.ToString(),
            Payload   = JsonSerializer.SerializeToElement(message),
        });

        await _redis.GetSubscriber()
            .PublishAsync(RedisChannel.Literal(channel), envelope);
    }
}

public sealed record RealtimeEnvelope
{
    public required string Channel { get; init; }
    public required string EventName { get; init; }
    public required RealtimeTarget Target { get; init; }
    public required string TargetId { get; init; }
    public required JsonElement Payload { get; init; }
}

public enum RealtimeTarget { User, Brand, Tenant, User }

RealtimeNotificationRelay (HostedService в user-api)​

// Pin.User.Api/Realtime/RealtimeNotificationRelay.cs

public sealed class RealtimeNotificationRelay : BackgroundService
{
    private readonly IConnectionMultiplexer _redis;
    private readonly IHubContext<UserNotificationHub, IUserNotificationClient> _hubContext;
    private readonly ILogger<RealtimeNotificationRelay> _logger;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var sub = _redis.GetSubscriber();

        // Подписываемся на все user и brand channels через pattern
        await sub.SubscribeAsync(
            RedisChannel.Pattern("Pin:realtime:*"),
            async (channel, value) =>
            {
                try
                {
                    var envelope = JsonSerializer.Deserialize<RealtimeEnvelope>(value.ToString());
                    if (envelope is null) return;

                    await DispatchToHub(envelope, stoppingToken);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error dispatching realtime message on channel {Channel}", channel);
                }
            });

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

    private async Task DispatchToHub(RealtimeEnvelope envelope, CancellationToken ct)
    {
        switch (envelope.Target)
        {
            case RealtimeTarget.User:
                await _hubContext.Clients
                    .Group($"user:{envelope.TargetId}")
                    .SendCoreAsync(envelope.EventName, [envelope.Payload], ct);
                break;

            case RealtimeTarget.Brand:
                await _hubContext.Clients
                    .Group($"brand:{envelope.TargetId}")
                    .SendCoreAsync(envelope.EventName, [envelope.Payload], ct);
                break;
        }
    }
}

Правила​

  • Redis backplane обязателен в staging и production (k8s multi-replica).
  • Локально (Docker Compose, один pod) backplane опционален — конфигурируется через REALTIME_BACKPLANE_ENABLED.
  • Каждое Hub-сообщение содержит только безопасные, non-PII данные (Id, статус, сумма).
  • Realtime не является source of truth — клиент должен перезапросить REST API при reconnect.
  • IRealtimeProducer используется в worker/background; IRealtimeNotificationPublisher — только в сервисе с Hub.
  • Rate limit: не более 10 push-сообщений в секунду на одного игрока.
  • Все push проходят через IRealtimeProducer или IRealtimeNotificationPublisher — прямое использование IHubContext в бизнес-коде запрещено.
  • Тип сообщения (eventName) — snake_case константа из RealtimeEvents.

Ссылки​

Обратные ссылки​

Автоссылки: где используется этот документ.