w4c-workflows-api/Services/Messaging/RedisStreamsTransport.cs

using Microsoft.Extensions.Configuration;
using StackExchange.Redis;

namespace w4c_workflows.Services.Messaging;

/// <summary>
/// Redis Streams implementation of <see cref="IJobQueue"/> and <see cref="IEventBus"/>.
///
/// Streams per tenant:
///   - wf:{tenant}:jobs    — task.run dispatch (consumer group `workers`)
///   - wf:{tenant}:events  — default handler-mode event stream
///   - wf:{tenant}:dlq     — dead-lettered messages
///
/// Delivery is at-least-once; the engine enforces idempotency via
/// correlation_id + a per-task idempotency key.
/// </summary>
public class RedisStreamsTransport : IJobQueue, IEventBus
{
    private readonly IDatabase _db;
    private readonly string _jobsTemplate;
    private readonly string _eventsTemplate;
    private readonly string _dlqTemplate;
    private readonly string _group;
    private readonly int _maxStreamLength;

    public RedisStreamsTransport(IConnectionMultiplexer redis, IConfiguration config)
    {
        _db = redis.GetDatabase();
        _jobsTemplate = config["Workflows:WorkerStream"] ?? "wf:{tenant}:jobs";
        _eventsTemplate = config["Workflows:EventStream"] ?? "wf:{tenant}:events";
        _dlqTemplate = config["Workflows:DlqStream"] ?? "wf:{tenant}:dlq";
        _group = config["Workflows:ConsumerGroup"] ?? "workers";
        // Bound every stream so jobs/events/results/dlq cannot grow without limit
        // on a long-lived Redis. 0 disables trimming. Approximate MAXLEN is cheap
        // (removes whole radix nodes), so it does not add meaningful latency.
        _maxStreamLength = config.GetValue("Workflows:StreamMaxLength", 50_000);
    }

    private string JobsStream(string tenantId) => _jobsTemplate.Replace("{tenant}", tenantId);
    private string EventStream(string tenantId, string stream) => Resolve(stream, tenantId);
    private string DlqStream(string tenantId) => _dlqTemplate.Replace("{tenant}", tenantId);

    private static string Resolve(string stream, string tenantId)
        => stream.Replace("{tenant}", tenantId);

    // ------------------------------------------------------------------ IJobQueue

    public Task EnsureGroupAsync(string tenantId, CancellationToken ct)
        => EnsureGroupCoreAsync(JobsStream(tenantId), ct);

    public Task<string> EnqueueAsync(string tenantId, IReadOnlyDictionary<string, string> fields, CancellationToken ct)
        => AddAsync(JobsStream(tenantId), fields);

    public Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string consumer, int count, CancellationToken ct)
        => ReadGroupCoreAsync(JobsStream(tenantId), consumer, count);

    public Task<IReadOnlyList<StreamMessage>> ClaimPendingAsync(string tenantId, string consumer, TimeSpan minIdle, int count, CancellationToken ct)
        => ClaimPendingCoreAsync(JobsStream(tenantId), consumer, minIdle, count);

    public Task<long> AckAsync(string tenantId, string messageId, CancellationToken ct)
        => _db.StreamAcknowledgeAsync(JobsStream(tenantId), _group, messageId);

    public Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
        => DeadLetterCoreAsync(JobsStream(tenantId), DlqStream(tenantId), messageId, fields, reason);

    public Task PublishDeadLetterAsync(string tenantId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
        => AddAsync(DlqStream(tenantId), DeadLetterFields(fields, reason, sourceId: null));

    // ------------------------------------------------------------------ IEventBus

    public Task<string> PublishAsync(string tenantId, string stream, IReadOnlyDictionary<string, string> fields, CancellationToken ct)
        => AddAsync(EventStream(tenantId, stream), fields);

    public Task EnsureGroupAsync(string tenantId, string stream, CancellationToken ct)
        => EnsureGroupCoreAsync(EventStream(tenantId, stream), ct);

    public Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string stream, string consumer, int count, CancellationToken ct)
        => ReadGroupCoreAsync(EventStream(tenantId, stream), consumer, count);

    public Task<IReadOnlyList<StreamMessage>> ClaimPendingAsync(string tenantId, string stream, string consumer, TimeSpan minIdle, int count, CancellationToken ct)
        => ClaimPendingCoreAsync(EventStream(tenantId, stream), consumer, minIdle, count);

    public Task<long> AckAsync(string tenantId, string stream, string messageId, CancellationToken ct)
        => _db.StreamAcknowledgeAsync(EventStream(tenantId, stream), _group, messageId);

    // ------------------------------------------------------------------ core

    private async Task<string> AddAsync(string key, IReadOnlyDictionary<string, string> fields)
    {
        var entries = fields.Select(kv => new NameValueEntry(kv.Key, kv.Value)).ToArray();
        var id = _maxStreamLength > 0
            ? await _db.StreamAddAsync(key, entries, _maxStreamLength, useApproximateMaxLength: true)
            : await _db.StreamAddAsync(key, entries);
        return id.ToString();
    }

    /// <summary>
    /// Creates the consumer group idempotently. The group starts at new messages
    /// (`&gt;`) so a fresh worker never replays stale history; redelivery of
    /// in-flight messages is handled by <see cref="ClaimPendingCoreAsync"/>.
    /// </summary>
    private async Task EnsureGroupCoreAsync(string key, CancellationToken ct)
    {
        try
        {
            await _db.StreamCreateConsumerGroupAsync(key, _group, StreamPosition.NewMessages, createStream: true);
        }
        catch (RedisServerException ex) when (ex.Message.Contains("BUSYGROUP", StringComparison.OrdinalIgnoreCase))
        {
            // Group already exists — fine.
        }
    }

    private async Task<IReadOnlyList<StreamMessage>> ReadGroupCoreAsync(string key, string consumer, int count)
    {
        var entries = await _db.StreamReadGroupAsync(key, _group, consumer, StreamPosition.NewMessages, count);
        return entries.Select(ToMessage).ToList();
    }

    private async Task<IReadOnlyList<StreamMessage>> ClaimPendingCoreAsync(string key, string consumer, TimeSpan minIdle, int count)
    {
        try
        {
            var result = await _db.StreamAutoClaimAsync(
                key, _group, consumer, (long)minIdle.TotalMilliseconds, StreamPosition.Beginning, count);
            return result.ClaimedEntries.Select(ToMessage).ToList();
        }
        catch (RedisServerException)
        {
            // Consumer group may not exist yet, or the stream is empty.
            return Array.Empty<StreamMessage>();
        }
    }

    private async Task DeadLetterCoreAsync(string key, string dlq, string messageId,
        IReadOnlyDictionary<string, string> fields, string reason)
    {
        try
        {
            await AddAsync(dlq, DeadLetterFields(fields, reason, messageId));
        }
        finally
        {
            // Ack the original so it leaves the pending list regardless of whether
            // the DLQ write succeeded (the failure is already captured in logs).
            await _db.StreamAcknowledgeAsync(key, _group, messageId);
        }
    }

    private static Dictionary<string, string> DeadLetterFields(
        IReadOnlyDictionary<string, string> fields, string reason, string? sourceId)
    {
        var dlqFields = new Dictionary<string, string>(fields)
        {
            ["dlq.reason"] = reason,
            ["dlq.at"] = DateTime.UtcNow.ToString("O"),
        };
        if (sourceId != null)
            dlqFields["dlq.source_id"] = sourceId;
        return dlqFields;
    }

    private static StreamMessage ToMessage(StreamEntry entry)
    {
        var dict = new Dictionary<string, string>(entry.Values.Length);
        foreach (var value in entry.Values)
            dict[value.Name.ToString()] = value.Value.ToString();
        return new StreamMessage(entry.Id.ToString(), dict);
    }
}