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

using System.Collections.Concurrent;
using System.Threading.Channels;

namespace w4c_workflows.Services.Messaging;

/// <summary>
/// In-process channel-based transport for <c>IJobQueue</c> and <c>IEventBus</c>.
/// Used in lite/self-hosted mode where Redis is not available. Store-backed
/// (ConcurrentDictionary of channels per stream), survives thread-boundary
/// handoffs but NOT process restarts. For self-hosted single-process deployment
/// this is acceptable (restart = redeliver pending jobs from DB state).
/// </summary>
public sealed class InMemoryTransport : IJobQueue, IEventBus
{
    private readonly ConcurrentDictionary<string, Channel<StreamMessage>> _channels = new();
    private long _idSeq;

    // ---- IJobQueue ----

    public Task EnsureGroupAsync(string tenantId, CancellationToken ct)
        => Task.CompletedTask; // No-op: channel consumers are implicit.

    public Task<string> EnqueueAsync(string tenantId, IReadOnlyDictionary<string, string> fields, CancellationToken ct)
    {
        var stream = $"wf:{tenantId}:jobs";
        var id = $"0-{Interlocked.Increment(ref _idSeq)}";
        var msg = new StreamMessage(id, fields);
        GetOrAddChannel(stream).Writer.TryWrite(msg);
        return Task.FromResult(id);
    }

    public Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string consumer, int count, CancellationToken ct)
        => DrainAsync($"wf:{tenantId}:jobs", count, ct);

    public Task<IReadOnlyList<StreamMessage>> ClaimPendingAsync(string tenantId, string consumer, TimeSpan minIdle, int count, CancellationToken ct)
        => Task.FromResult<IReadOnlyList<StreamMessage>>(Array.Empty<StreamMessage>()); // No pending in memory.

    public Task<long> AckAsync(string tenantId, string messageId, CancellationToken ct)
        => Task.FromResult(1L); // Ack is a no-op in memory.

    public Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
    {
        var dlq = $"wf:{tenantId}:dlq";
        var id = $"0-{Interlocked.Increment(ref _idSeq)}";
        var dlqFields = new Dictionary<string, string>(fields) { ["deadLetterReason"] = reason };
        GetOrAddChannel(dlq).Writer.TryWrite(new StreamMessage(id, dlqFields));
        return Task.CompletedTask;
    }

    public Task PublishDeadLetterAsync(string tenantId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
    {
        var dlq = $"wf:{tenantId}:dlq";
        var id = $"0-{Interlocked.Increment(ref _idSeq)}";
        var dlqFields = new Dictionary<string, string>(fields) { ["deadLetterReason"] = reason };
        GetOrAddChannel(dlq).Writer.TryWrite(new StreamMessage(id, dlqFields));
        return Task.CompletedTask;
    }

    // ---- IEventBus ----

    public Task<string> PublishAsync(string tenantId, string stream, IReadOnlyDictionary<string, string> fields, CancellationToken ct)
    {
        var id = $"0-{Interlocked.Increment(ref _idSeq)}";
        var msg = new StreamMessage(id, fields);
        // Expand {tenant} placeholder.
        var resolved = stream.Replace("{tenant}", tenantId);
        GetOrAddChannel(resolved).Writer.TryWrite(msg);
        return Task.FromResult(id);
    }

    Task IEventBus.EnsureGroupAsync(string tenantId, string stream, CancellationToken ct)
        => Task.CompletedTask;

    Task<IReadOnlyList<StreamMessage>> IEventBus.ReadGroupAsync(string tenantId, string stream, string consumer, int count, CancellationToken ct)
    {
        var resolved = stream.Replace("{tenant}", tenantId);
        return DrainAsync(resolved, count, ct);
    }

    Task<IReadOnlyList<StreamMessage>> IEventBus.ClaimPendingAsync(string tenantId, string stream, string consumer, TimeSpan minIdle, int count, CancellationToken ct)
        => Task.FromResult<IReadOnlyList<StreamMessage>>(Array.Empty<StreamMessage>());

    Task<long> IEventBus.AckAsync(string tenantId, string stream, string messageId, CancellationToken ct)
        => Task.FromResult(1L);

    // ---- Helpers ----

    private Channel<StreamMessage> GetOrAddChannel(string key)
        => _channels.GetOrAdd(key, _ => Channel.CreateBounded<StreamMessage>(
            new BoundedChannelOptions(ChannelCapacity)
            {
                SingleReader = false,
                SingleWriter = false,
                // Bounded so a stalled consumer cannot grow memory without limit.
                // Under pressure the OLDEST message is dropped: lite mode is
                // single-process and jobs/notifications are recoverable from DB
                // state, so dropping the stale tail is preferable to OOM.
                FullMode = BoundedChannelFullMode.DropOldest,
            }));

    /// <summary>Per-stream in-memory backlog ceiling.</summary>
    private const int ChannelCapacity = 10_000;

    private async Task<IReadOnlyList<StreamMessage>> DrainAsync(string stream, int count, CancellationToken ct)
    {
        if (!_channels.TryGetValue(stream, out var ch))
            return Array.Empty<StreamMessage>();

        var result = new List<StreamMessage>(count);
        while (result.Count < count)
        {
            try
            {
                if (ch.Reader.TryRead(out var msg))
                {
                    result.Add(msg);
                }
                else
                {
                    // Wait briefly for a message.
                    using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
                    cts.CancelAfter(TimeSpan.FromMilliseconds(50));
                    try
                    {
                        var read = await ch.Reader.ReadAsync(cts.Token);
                        result.Add(read);
                    }
                    catch (OperationCanceledException) when (!ct.IsCancellationRequested)
                    {
                        break; // No more messages available right now.
                    }
                }
            }
            catch (OperationCanceledException)
            {
                break;
            }
        }
        return result;
    }
}