src/RedisLab.Infrastructure/Notifications/NotificationStreamConsumerHostedService.cs - master - afanasevn/RedisLab

src/RedisLab.Infrastructure/Notifications/NotificationStreamConsumerHostedService.cs
169 строк6 KB
IBS\NAfanasev
Solution commit
24 июн 2026, 16:45
24 июн 2026, 16:45bab80b7
using System.Text.Json;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using RedisLab.Application.Abstractions;
using RedisLab.Application.Contracts.Activity;
using RedisLab.Application.Contracts.Notifications;
using RedisLab.Application.Redis;
using RedisLab.Infrastructure.Hosting;
using StackExchange.Redis;

namespace RedisLab.Infrastructure.Notifications;

/// <summary>
/// Consumer group notifier: XREADGROUP => обработка => XACK.
/// После restart читает необработанный backlog (в отличие от Pub/Sub activity:feed).
/// </summary>
public sealed class NotificationStreamConsumerHostedService(
    IConnectionMultiplexer redis,
    InstanceIdentity instanceIdentity,
    IServiceScopeFactory scopeFactory,
    ILogger<NotificationStreamConsumerHostedService> logger) : BackgroundService
{
    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
    };

    private readonly string _consumerName = $"notifier-{instanceIdentity.Id}";

    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await EnsureConsumerGroupAsync(stoppingToken);

        logger.LogInformation(
            "Notification stream consumer {Consumer} в группе {Group}, stream {Stream}",
            _consumerName,
            RedisKeyNames.NotificationsConsumerGroup,
            RedisKeyNames.NotificationsStream);

        var database = redis.GetDatabase();

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // ">" — только новые для группы записи; непрочитанный backlog доставляется после restart API.
                var entries = await database.StreamReadGroupAsync(
                    RedisKeyNames.NotificationsStream,
                    RedisKeyNames.NotificationsConsumerGroup,
                    _consumerName,
                    ">",
                    count: 10);

                if (entries.Length == 0)
                {
                    await Task.Delay(TimeSpan.FromSeconds(2), stoppingToken);
                    continue;
                }

                foreach (var entry in entries)
                {
                    await ProcessEntryAsync(database, entry, stoppingToken);
                }
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "Ошибка чтения notifications:stream");
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
            }
        }
    }

    /// <summary>
    /// Создаёт consumer group при первом запуске (MKSTREAM если потока ещё нет).
    /// </summary>
    private async Task EnsureConsumerGroupAsync(CancellationToken ct)
    {
        var database = redis.GetDatabase();

        try
        {
            await database.StreamCreateConsumerGroupAsync(
                RedisKeyNames.NotificationsStream,
                RedisKeyNames.NotificationsConsumerGroup,
                "0-0",
                createStream: true);
        }
        catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP", StringComparison.Ordinal))
        {
            // Группа уже существует — нормально при повторном старте.
        }
    }

    /// <summary>
    /// Десериализует payload, пишет в in-memory store и подтверждает XACK.
    /// </summary>
    private async Task ProcessEntryAsync(
        IDatabase database,
        StreamEntry entry,
        CancellationToken ct)
    {
        var payloadValue = entry.Values.FirstOrDefault(x => x.Name == "payload").Value;
        if (payloadValue.IsNullOrEmpty)
        {
            logger.LogWarning("Stream entry {Id} без поля payload, XACK", entry.Id);
            await database.StreamAcknowledgeAsync(
                RedisKeyNames.NotificationsStream,
                RedisKeyNames.NotificationsConsumerGroup,
                entry.Id);
            return;
        }

        ActivityFeedMessage? message;
        try
        {
            message = JsonSerializer.Deserialize<ActivityFeedMessage>(payloadValue.ToString(), JsonOptions);
        }
        catch (JsonException ex)
        {
            logger.LogError(ex, "Невалидный JSON в stream entry {Id}", entry.Id);
            await database.StreamAcknowledgeAsync(
                RedisKeyNames.NotificationsStream,
                RedisKeyNames.NotificationsConsumerGroup,
                entry.Id);
            return;
        }

        if (message is null)
        {
            await database.StreamAcknowledgeAsync(
                RedisKeyNames.NotificationsStream,
                RedisKeyNames.NotificationsConsumerGroup,
                entry.Id);
            return;
        }

        var processedAt = DateTime.UtcNow;
        var dto = new ProcessedNotificationDto(
            entry.Id.ToString(),
            message.Type,
            message.TaskId,
            message.Message,
            message.At,
            processedAt);

        using (var scope = scopeFactory.CreateScope())
        {
            var store = scope.ServiceProvider.GetRequiredService<IProcessedNotificationStore>();
            store.Add(dto);
        }

        await database.StreamAcknowledgeAsync(
            RedisKeyNames.NotificationsStream,
            RedisKeyNames.NotificationsConsumerGroup,
            entry.Id);

        logger.LogInformation(
            "Notification XACK {StreamId} type={Type} (consumer {Consumer})",
            entry.Id,
            message.Type,
            _consumerName);
    }
}