w4c-workflows-api/w4c-workflows-api.Tests/Fakes.cs

using w4c_workflows.Services;
using w4c_workflows.Services.Messaging;

namespace w4c_workflows.Tests;

/// <summary>In-memory <see cref="IJobQueue"/> capturing enqueued <c>task.run</c> fields.</summary>
internal sealed class FakeJobQueue : IJobQueue
{
    public List<(string TenantId, IReadOnlyDictionary<string, string> Fields)> Enqueued { get; } = new();
    public List<(string TenantId, IReadOnlyDictionary<string, string> Fields, string Reason)> DeadLettered { get; } = new();

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

    public Task<string> EnqueueAsync(string tenantId, IReadOnlyDictionary<string, string> fields, CancellationToken ct)
    {
        Enqueued.Add((tenantId, fields));
        return Task.FromResult(Enqueued.Count.ToString());
    }

    public Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string consumer, int count, CancellationToken ct)
        => Task.FromResult<IReadOnlyList<StreamMessage>>(Array.Empty<StreamMessage>());

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

    public Task<long> AckAsync(string tenantId, string messageId, CancellationToken ct) => Task.FromResult(0L);

    public Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
    {
        DeadLettered.Add((tenantId, fields, reason));
        return Task.CompletedTask;
    }

    public Task PublishDeadLetterAsync(string tenantId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct)
    {
        DeadLettered.Add((tenantId, fields, reason));
        return Task.CompletedTask;
    }

    public List<IReadOnlyDictionary<string, string>> EnqueuedFor(string tenantId)
        => Enqueued.Where(x => x.TenantId == tenantId).Select(x => x.Fields).ToList();

    public List<(IReadOnlyDictionary<string, string> Fields, string Reason)> DeadLetteredFor(string tenantId)
        => DeadLettered.Where(x => x.TenantId == tenantId).Select(x => (x.Fields, x.Reason)).ToList();
}

/// <summary>
/// In-memory <see cref="IEventBus"/>. Stream names resolve the <c>{tenant}</c>
/// placeholder exactly like the real Redis transport; tests seed result messages
/// with <see cref="SeedResultAsync"/> and the engine reads + acks them.
/// </summary>
internal sealed class FakeEventBus : IEventBus
{
    private readonly Dictionary<string, Queue<StreamMessage>> _messages = new();
    private int _nextId = 1;

    public List<string> Acked { get; } = new();
    public List<(string TenantId, string Stream, IReadOnlyDictionary<string, string> Fields)> Published { get; } = new();

    public void SeedResult(string tenantId, IReadOnlyDictionary<string, string> fields)
        => Seed(Streams.Results.Replace("{tenant}", tenantId), fields);

    public void Seed(string streamKey, IReadOnlyDictionary<string, string> fields)
    {
        if (!_messages.TryGetValue(streamKey, out var queue))
        {
            queue = new Queue<StreamMessage>();
            _messages[streamKey] = queue;
        }
        queue.Enqueue(new StreamMessage((_nextId++).ToString(), fields));
    }

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

    public Task<string> PublishAsync(string tenantId, string stream, IReadOnlyDictionary<string, string> fields, CancellationToken ct)
    {
        Published.Add((tenantId, stream, fields));
        return Task.FromResult((_nextId++).ToString());
    }

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

    public Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string stream, string consumer, int count, CancellationToken ct)
    {
        var key = Resolve(stream, tenantId);
        var list = new List<StreamMessage>();
        if (_messages.TryGetValue(key, out var queue))
            while (list.Count < count && queue.Count > 0)
                list.Add(queue.Dequeue());
        return Task.FromResult<IReadOnlyList<StreamMessage>>(list);
    }

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

    public Task<long> AckAsync(string tenantId, string stream, string messageId, CancellationToken ct)
    {
        Acked.Add(messageId);
        return Task.FromResult(1L);
    }
}

/// <summary>In-memory <see cref="ILeaseService"/> keyed by tenant + run id.</summary>
internal sealed class FakeLeaseService : ILeaseService
{
    private readonly Dictionary<string, string> _held = new();

    public Task<bool> AcquireAsync(string tenantId, string runId, string owner, TimeSpan ttl)
    {
        var key = $"{tenantId}:{runId}";
        if (_held.ContainsKey(key))
            return Task.FromResult(false);
        _held[key] = owner;
        return Task.FromResult(true);
    }

    public Task<bool> RenewAsync(string tenantId, string runId, string owner, TimeSpan ttl)
        => Task.FromResult(_held.TryGetValue($"{tenantId}:{runId}", out var held) && held == owner);

    public Task<bool> ReleaseAsync(string tenantId, string runId, string owner)
    {
        var key = $"{tenantId}:{runId}";
        if (_held.TryGetValue(key, out var held) && held == owner)
        {
            _held.Remove(key);
            return Task.FromResult(true);
        }
        return Task.FromResult(false);
    }
}