68 lines
3.2 KiB
C#
68 lines
3.2 KiB
C#
namespace w4c_workflows.Services.Messaging;
|
|
|
|
/// <summary>A single entry pulled off a Redis stream / (later) RabbitMQ queue.</summary>
|
|
public sealed record StreamMessage(string Id, IReadOnlyDictionary<string, string> Fields);
|
|
|
|
/// <summary>
|
|
/// Default event stream for <c>handler</c>-mode subscriptions when the YAML
|
|
/// trigger does not name an explicit <c>stream</c>.
|
|
/// </summary>
|
|
public static class Streams
|
|
{
|
|
public const string DefaultEvents = "wf:{tenant}:events";
|
|
|
|
/// <summary>Worker → control-plane result events (<c>task.result</c>).</summary>
|
|
public const string Results = "wf:{tenant}:results";
|
|
}
|
|
|
|
/// <summary>
|
|
/// Job dispatch for <c>task.run</c> messages. The v1 implementation is Redis
|
|
/// Streams with consumer groups; a RabbitMQ transport can be swapped in behind
|
|
/// this interface without touching the engine.
|
|
/// </summary>
|
|
public interface IJobQueue
|
|
{
|
|
/// <summary>Ensures the tenant's jobs consumer group exists.</summary>
|
|
Task EnsureGroupAsync(string tenantId, CancellationToken ct);
|
|
|
|
/// <summary>Enqueues a job; returns the stream entry id.</summary>
|
|
Task<string> EnqueueAsync(string tenantId, IReadOnlyDictionary<string, string> fields, CancellationToken ct);
|
|
|
|
/// <summary>Reads new (undelivered) jobs for a consumer.</summary>
|
|
Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string consumer, int count, CancellationToken ct);
|
|
|
|
/// <summary>Claims jobs pending for longer than <paramref name="minIdle"/> (redelivery).</summary>
|
|
Task<IReadOnlyList<StreamMessage>> ClaimPendingAsync(string tenantId, string consumer, TimeSpan minIdle, int count, CancellationToken ct);
|
|
|
|
Task<long> AckAsync(string tenantId, string messageId, CancellationToken ct);
|
|
|
|
/// <summary>Copies a failed job to the DLQ and acks it out of the group.</summary>
|
|
Task DeadLetterAsync(string tenantId, string messageId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct);
|
|
|
|
/// <summary>
|
|
/// Writes a dead-letter entry without a source job to ack. Used by the
|
|
/// control-plane saga path: a task that exhausted retries (or a failed
|
|
/// compensation) has no live jobs-stream entry left to acknowledge.
|
|
/// </summary>
|
|
Task PublishDeadLetterAsync(string tenantId, IReadOnlyDictionary<string, string> fields, string reason, CancellationToken ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Event publish/subscribe (Redis Streams now, RabbitMQ later). A
|
|
/// <c>handler</c>-mode workflow subscribes to a named stream via
|
|
/// <c>trigger.stream</c>; the trigger engine consumes it through a consumer
|
|
/// group and runs the downstream <c>nextId</c> chain per event.
|
|
/// </summary>
|
|
public interface IEventBus
|
|
{
|
|
Task<string> PublishAsync(string tenantId, string stream, IReadOnlyDictionary<string, string> fields, CancellationToken ct);
|
|
|
|
Task EnsureGroupAsync(string tenantId, string stream, CancellationToken ct);
|
|
|
|
Task<IReadOnlyList<StreamMessage>> ReadGroupAsync(string tenantId, string stream, string consumer, int count, CancellationToken ct);
|
|
|
|
Task<IReadOnlyList<StreamMessage>> ClaimPendingAsync(string tenantId, string stream, string consumer, TimeSpan minIdle, int count, CancellationToken ct);
|
|
|
|
Task<long> AckAsync(string tenantId, string stream, string messageId, CancellationToken ct);
|
|
}
|