using MassTransit; using MassTransit.Courier.Contracts; using Musify.Infrastructure.MassTransit.Activities.Files; using Musify.Infrastructure.MassTransit.Arguments; using Musify.Infrastructure.MassTransit.Consumers; namespace Musify.Infrastructure.MassTransit.RoutingSlip; /// Shared building blocks for every routing slip in the worker. internal static class RoutingSlips { public static Uri ExecuteUri(string endpointName) => new($"queue:{endpointName}_execute"); public static Uri ConsumerUri(string queueName) => new($"queue:{queueName}"); /// A new slip that carries the correlation id and always cleans its temporary directory up. public static RoutingSlipBuilder Create(Guid? correlationId) { var builder = new RoutingSlipBuilder(NewId.NextGuid()); builder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty); builder.AddSubscription( ConsumerUri(RoutingSlipCleanUpConsumer.QueueName), RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted); return builder; } public static RoutingSlipBuilder AddStep( this RoutingSlipBuilder builder, string name, string executeEndpointName, object arguments) { builder.AddActivity(name, ExecuteUri(executeEndpointName), arguments); return builder; } /// Publishes a failure event for the subject when the slip faults. public static RoutingSlipBuilder TrackFaults( this RoutingSlipBuilder builder, Guid subjectId, string processKind) { builder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, subjectId); builder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind); builder.AddSubscription(ConsumerUri(ProcessingSlipFaultConsumer.QueueName), RoutingSlipEvents.Faulted); return builder; } public static RoutingSlipBuilder AddRemoveFile( this RoutingSlipBuilder builder, string stepName, string bucket, string key) => builder.AddStep(stepName, RemoveFileFromBucketActivity.ExecuteEndpointName, new RemoveFileFromBucketArguments(bucket, key)); }