src/RedisLab.Infrastructure/Activity/ActivityFeedService.cs - master - afanasevn/RedisLab

src/RedisLab.Infrastructure/Activity/ActivityFeedService.cs
84 строки3 KB
IBS\NAfanasev
Solution commit
24 июн 2026, 16:45
24 июн 2026, 16:45bab80b7
using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using RedisLab.Application.Abstractions;
using RedisLab.Application.Contracts.Activity;
using RedisLab.Application.Redis;
using RedisLab.Domain.Entities;
using RedisLab.Infrastructure.Persistence;
using StackExchange.Redis;

namespace RedisLab.Infrastructure.Activity;

/// <summary>
/// Live-лента: PostgreSQL (durability) + Redis Pub/Sub (real-time fan-out).
/// </summary>
public sealed class ActivityFeedService(
    AppDbContext db,
    IConnectionMultiplexer redis,
    INotificationStreamPublisher notificationStreamPublisher,
    ILogger<ActivityFeedService> logger) : IActivityFeedService
{
    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
    };

    /// <inheritdoc />
    public async Task PublishAsync(
        ActivityFeedMessage message,
        CancellationToken ct)
    {
        var entity = new ActivityEvent
        {
            Id = Guid.NewGuid(),
            EventType = message.Type,
            TaskId = message.TaskId,
            Message = message.Message,
            OccurredAt = message.At,
        };

        db.ActivityEvents.Add(entity);
        await db.SaveChangesAsync(ct);

        var payload = JsonSerializer.Serialize(message, JsonOptions);

        // Pub/Sub — fire-and-forget: подписчики offline не получат сообщение после reconnect.
        // Поэтому сначала пишем в PG, а Redis используем только для push живым клиентам.
        var subscriber = redis.GetSubscriber();
        var receivers = await subscriber.PublishAsync(
            RedisChannel.Literal(RedisKeyNames.ActivityFeedChannel),
            payload);

        logger.LogInformation(
            "Activity PUBLISH {Channel}: type={Type}, taskId={TaskId}, receivers={Receivers}",
            RedisKeyNames.ActivityFeedChannel,
            message.Type,
            message.TaskId,
            receivers);

        // Stream: durable log — consumer прочитает backlog после offline/restart.
        // Pub/Sub выше — только live fan-out; Stream — учебный outbox-lite, не production bus.
        await notificationStreamPublisher.EnqueueAsync(message, ct);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<ActivityEventDto>> GetRecentAsync(
        int count,
        CancellationToken ct)
    {
        var take = Math.Clamp(count, 1, 100);

        return await db.ActivityEvents
            .AsNoTracking()
            .OrderByDescending(x => x.OccurredAt)
            .Take(take)
            .Select(x => new ActivityEventDto(
                x.Id,
                x.EventType,
                x.TaskId,
                x.Message,
                x.OccurredAt))
            .ToListAsync(ct);
    }
}