Deduplicate MassTransit routing slips, activities and consumers

This commit is contained in:
2026-09-24 12:25:55 +02:00
parent 0cba32f776
commit 405df082cc
76 changed files with 922 additions and 1933 deletions
@@ -34,6 +34,8 @@ public interface IDatabase
public Task<IDatabaseTransaction> BeginTransactionAsync(System.Data.IsolationLevel isolationLevel, CancellationToken cancellationToken);
public DbSet<TEntity> Set<TEntity>() where TEntity : class;
public Task<int> SaveChangesAsync(CancellationToken cancellationToken);
public Task<ErrorOr<Success>> TrySaveChangesAsync(Error onUniqueViolation, CancellationToken cancellationToken);
@@ -1,35 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Albums;
internal class DeleteAlbumFromDbActivity(
IDatabase database,
ILogger<DeleteAlbumFromDbActivity> logger)
: IExecuteActivity<DeleteAlbumFromDbArguments>
{
public const string ExecuteEndpointName = "delete-album-from-db";
public async Task<ExecutionResult> Execute(ExecuteContext<DeleteAlbumFromDbArguments> executeContext)
{
var album = await database.Albums
.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.AlbumId, executeContext.CancellationToken);
if (album == null)
{
logger.LogInformation(
"Album {AlbumId} not found while deleting from DB, skipping",
executeContext.Arguments.AlbumId);
return executeContext.Completed();
}
database.Albums.Remove(album);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Deleted album {AlbumId} from DB", album.Id);
return executeContext.Completed();
}
}
@@ -1,31 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Albums;
internal class MarkAlbumAsFailedActivity(
IDatabase database,
ILogger<MarkAlbumAsFailedActivity> logger)
: IExecuteActivity<MarkAlbumAsFailedArguments>
{
public const string ExecuteEndpointName = "mark-album-as-failed";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkAlbumAsFailedArguments> executeContext)
{
var album = await database.Albums
.SingleOrDefaultAsync(a => a.Id == executeContext.Arguments.AlbumId, executeContext.CancellationToken);
if (album != null)
{
album.LifeCycleStatus = LifeCycleStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked album {AlbumId} as failed", executeContext.Arguments.AlbumId);
}
return executeContext.Completed();
}
}
@@ -1,36 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Albums;
internal class MarkAlbumAsRemovingActivity(
IDatabase database,
ILogger<MarkAlbumAsRemovingActivity> logger)
: IExecuteActivity<MarkAlbumAsRemovingArguments>
{
public const string ExecuteEndpointName = "mark-album-as-removing";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkAlbumAsRemovingArguments> executeContext)
{
var album = await database.Albums
.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.AlbumId, executeContext.CancellationToken);
if (album == null)
{
logger.LogInformation(
"Album {AlbumId} not found while marking as removing, skipping",
executeContext.Arguments.AlbumId);
return executeContext.Completed();
}
album.LifeCycleStatus = LifeCycleStatus.Removing;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked album {AlbumId} as removing", album.Id);
return executeContext.Completed();
}
}
@@ -1,10 +1,9 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.Entities;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Activities.Audio;
@@ -17,26 +16,15 @@ internal class GenerateAudioWorkflowPathsActivity(
public async Task<ExecutionResult> Execute(ExecuteContext<GenerateAudioWorkflowPathsArguments> executeContext)
{
Track track;
try
{
var getTrack = await database.Tracks.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.TrackId, executeContext.CancellationToken);
track = getTrack ?? throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found.");
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to retrieve track for TrackId {TrackId}",
executeContext.Arguments.TrackId);
throw;
}
var arguments = executeContext.Arguments;
var track = await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
try
{
track.Audio.TranscodeStatus = ProcessingStatus.Processing;
var folderName = Guid.NewGuid().ToString();
var sourceFileName = Path.GetFileName(executeContext.Arguments.SourceKey);
var workingDirectory = Path.Combine(executeContext.Arguments.TemporaryRootDirectory, folderName);
var sourceFilePath = Path.Combine(workingDirectory, sourceFileName);
var workingDirectory = Path.Combine(arguments.TemporaryRootDirectory, Guid.NewGuid().ToString());
var sourceFilePath = Path.Combine(workingDirectory, Path.GetFileName(arguments.SourceKey));
Directory.CreateDirectory(workingDirectory);
await database.SaveChangesAsync(executeContext.CancellationToken);
@@ -50,22 +38,9 @@ internal class GenerateAudioWorkflowPathsActivity(
[RoutingSlipVariableNames.Audio.TranscodedDirectory] = workingDirectory
});
}
catch (Exception exception)
catch
{
logger.LogError(exception, "Failed to generate audio workflow paths for {SourceKey}",
executeContext.Arguments.SourceKey);
try
{
track.Audio.TranscodeStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
}
catch (Exception dbException)
{
logger.LogError(dbException, "Failed to update ProcessingStatus to Failed for track {TrackId}",
executeContext.Arguments.TrackId);
}
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
throw;
}
}
@@ -0,0 +1,24 @@
using Microsoft.EntityFrameworkCore;
using Musify.Application.Contracts;
using Musify.Domain.Entities;
using Musify.Domain.ValueObjects;
namespace Musify.Infrastructure.MassTransit.Activities.Audio;
/// <summary>Shared bookkeeping of the audio processing activities.</summary>
internal static class TrackAudioStatus
{
public static async Task<Track> LoadAsync(IDatabase database, Guid trackId, CancellationToken cancellationToken) =>
await database.Tracks.SingleOrDefaultAsync(track => track.Id == trackId, cancellationToken)
?? throw new InvalidOperationException($"Track with id {trackId} not found");
public static async Task MarkFailedAsync(IDatabase database, Guid trackId, CancellationToken cancellationToken)
{
var track = await database.Tracks.SingleOrDefaultAsync(t => t.Id == trackId, cancellationToken);
if (track == null)
return;
track.Audio.TranscodeStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(cancellationToken);
}
}
@@ -1,11 +1,9 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.Entities;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Activities.Audio;
@@ -19,22 +17,13 @@ internal class TranscodeAudioActivity(
public async Task<ExecutionResult> Execute(ExecuteContext<TranscodeAudioArguments> executeContext)
{
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
var workingDirectory = executeContext.GetVariable<string>(executeContext.Arguments.WorkingDirectoryVariable);
ArgumentNullException.ThrowIfNull(workingDirectory, nameof(workingDirectory));
var arguments = executeContext.Arguments;
var sourceFilePath = executeContext.GetVariable<string>(arguments.SourceFilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath);
var workingDirectory = executeContext.GetVariable<string>(arguments.WorkingDirectoryVariable);
ArgumentNullException.ThrowIfNull(workingDirectory);
Track track;
try
{
var getTrack = await database.Tracks.SingleOrDefaultAsync(t => t.Id == executeContext.Arguments.TrackId, executeContext.CancellationToken);
track = getTrack ?? throw new InvalidOperationException($"Track with Id {executeContext.Arguments.TrackId} not found");
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to retrieve track with Id {TrackId} from database", executeContext.Arguments.TrackId);
throw;
}
await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
try
{
@@ -42,9 +31,7 @@ internal class TranscodeAudioActivity(
await using (var fileStream = File.OpenRead(sourceFilePath))
{
transcodeResult = await audioTranscoder.TranscodeToAudioFileAsync(
fileStream,
workingDirectory,
executeContext.CancellationToken);
fileStream, workingDirectory, executeContext.CancellationToken);
}
if (transcodeResult.StatusCode != 0)
@@ -52,51 +39,27 @@ internal class TranscodeAudioActivity(
File.Delete(sourceFilePath);
var log = new TranscodeAudioLog(workingDirectory);
return executeContext.CompletedWithVariables(log, new Dictionary<string, object>
return executeContext.CompletedWithVariables(new TranscodeAudioLog(workingDirectory), new Dictionary<string, object>
{
[RoutingSlipVariableNames.Audio.DurationSeconds] = transcodeResult.Duration.TotalSeconds
});
}
catch (Exception exception)
catch
{
logger.LogError(exception, "Failed to transcode {SourceFilePath}", sourceFilePath);
try
{
track.Audio.TranscodeStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
}
catch (Exception dbException)
{
logger.LogError(dbException, "Failed to update ProcessingStatus to Failed for track {TrackId}",
executeContext.Arguments.TrackId);
}
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
throw;
}
}
public Task<CompensationResult> Compensate(CompensateContext<TranscodeAudioLog> compensateContext)
{
try
{
if (Directory.Exists(compensateContext.Log.WorkingDirectory))
{
Directory.Delete(compensateContext.Log.WorkingDirectory, recursive: true);
}
var workingDirectory = compensateContext.Log.WorkingDirectory;
if (Directory.Exists(workingDirectory))
Directory.Delete(workingDirectory, recursive: true);
else
{
logger.LogWarning("Working directory {WorkingDirectory} not found during compensation",
compensateContext.Log.WorkingDirectory);
}
logger.LogWarning("Working directory {WorkingDirectory} not found during compensation", workingDirectory);
return Task.FromResult(compensateContext.Compensated());
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate transcoded audio directory {WorkingDirectory}", compensateContext.Log.WorkingDirectory);
return Task.FromResult(compensateContext.Failed(exception));
}
}
}
@@ -1,9 +1,7 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Application.Events;
using Musify.Domain.Entities;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs;
@@ -20,92 +18,49 @@ internal class UpdateTrackAudioActivity(
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateTrackAudioArguments> executeContext)
{
Track track;
var arguments = executeContext.Arguments;
var track = await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
try
{
var getTrack = await database.Tracks.SingleOrDefaultAsync(t => t.Id == executeContext.Arguments.TrackId, executeContext.CancellationToken);
track = getTrack ?? throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found");
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to retrieve track {TrackId}",
executeContext.Arguments.TrackId);
throw;
}
var audioFolderKey = executeContext.Arguments.AudioFolderKey;
logger.LogDebug("Resolved audio folder key: {AudioFolderKey}", audioFolderKey);
if (string.IsNullOrWhiteSpace(audioFolderKey))
{
if (string.IsNullOrWhiteSpace(arguments.AudioFolderKey))
throw new InvalidOperationException($"Audio folder key is empty for track {track.Id}");
}
try
{
var durationSeconds = executeContext.GetVariable<double>(executeContext.Arguments.DurationSecondsVariable)
?? throw new InvalidOperationException($"Missing routing slip variable {executeContext.Arguments.DurationSecondsVariable}");
var durationSeconds = executeContext.GetVariable<double>(arguments.DurationSecondsVariable)
?? throw new InvalidOperationException($"Missing routing slip variable {arguments.DurationSecondsVariable}");
var log = new UpdateTrackAudioLog(
track.Id,
track.Audio.FolderName,
track.DurationSeconds);
var log = new UpdateTrackAudioLog(track.Id, track.Audio.FolderName, track.DurationSeconds);
track.Audio.FolderName = Path.GetFileName(audioFolderKey);
track.Audio.FolderName = Path.GetFileName(arguments.AudioFolderKey);
track.DurationSeconds = durationSeconds;
track.Audio.TranscodeStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated track {TrackId} audio", executeContext.Arguments.TrackId);
logger.LogInformation("Updated track {TrackId} audio", arguments.TrackId);
await publishEndpoint.Publish(
new TrackAudioProcessed(executeContext.Arguments.TrackId),
executeContext.CancellationToken);
await publishEndpoint.Publish(new TrackAudioProcessed(arguments.TrackId), executeContext.CancellationToken);
return executeContext.Completed(log);
}
catch (Exception exception)
catch
{
logger.LogError(exception, "Failed to update track {TrackId} audio",
executeContext.Arguments.TrackId);
try
{
track.Audio.TranscodeStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
}
catch (Exception dbException)
{
logger.LogError(dbException, "Failed to update ProcessingStatus to Failed for track {TrackId}",
executeContext.Arguments.TrackId);
}
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
throw;
}
}
public async Task<CompensationResult> Compensate(CompensateContext<UpdateTrackAudioLog> compensateContext)
{
try
{
var track = await database.Tracks.SingleOrDefaultAsync(t => t.Id == compensateContext.Log.TrackId, compensateContext.CancellationToken);
if (track == null)
throw new InvalidOperationException($"Track with id {compensateContext.Log.TrackId} not found");
var log = compensateContext.Log;
track.Audio.FolderName = compensateContext.Log.PreviousAudioFolderName;
track.DurationSeconds = compensateContext.Log.PreviousDurationSeconds;
var track = await TrackAudioStatus.LoadAsync(database, log.TrackId, compensateContext.CancellationToken);
track.Audio.FolderName = log.PreviousAudioFolderName;
track.DurationSeconds = log.PreviousDurationSeconds;
track.Audio.TranscodeStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(compensateContext.CancellationToken);
logger.LogInformation("Compensated track {TrackId} audio", compensateContext.Log.TrackId);
logger.LogInformation("Compensated track {TrackId} audio", log.TrackId);
return compensateContext.Compensated();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate track {TrackId} audio",
compensateContext.Log.TrackId);
return compensateContext.Failed(exception);
}
}
}
@@ -7,7 +7,7 @@ using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Files;
internal class CopyFileInBucketActivity(
IStorageService storageHandler,
IStorageService storageService,
ILogger<CopyFileInBucketActivity> logger)
: IActivity<CopyFileInBucketArguments, CopyFileInBucketLog>
{
@@ -17,7 +17,7 @@ internal class CopyFileInBucketActivity(
{
try
{
await storageHandler.CopyFileAsync(
await storageService.CopyFileAsync(
executeContext.Arguments.SourceBucket,
executeContext.Arguments.SourceKey,
executeContext.Arguments.DestinationBucket,
@@ -43,7 +43,7 @@ internal class CopyFileInBucketActivity(
{
try
{
await storageHandler.RemoveFileAsync(
await storageService.RemoveFileAsync(
compensateContext.Log.DestinationBucket,
compensateContext.Log.DestinationKey,
compensateContext.CancellationToken);
@@ -7,7 +7,7 @@ using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Files;
internal class DownloadFileFromBucketActivity(
IStorageService storageHandler,
IStorageService storageService,
ILogger<DownloadFileFromBucketActivity> logger)
: IActivity<DownloadFileFromBucketArguments, DownloadFileFromBucketLog>
{
@@ -16,11 +16,11 @@ internal class DownloadFileFromBucketActivity(
public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext)
{
var destinationPath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
ArgumentNullException.ThrowIfNull(destinationPath, nameof(destinationPath));
ArgumentNullException.ThrowIfNull(destinationPath);
try
{
using var fileStream = await storageHandler.GetFileAsync(
using var fileStream = await storageService.GetFileAsync(
executeContext.Arguments.Bucket,
executeContext.Arguments.Key,
executeContext.CancellationToken)
@@ -8,7 +8,7 @@ using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Files;
internal class RemoveFileFromBucketActivity(
IStorageService storageHandler,
IStorageService storageService,
ILogger<RemoveFileFromBucketActivity> logger)
: IExecuteActivity<RemoveFileFromBucketArguments>
{
@@ -18,7 +18,7 @@ internal class RemoveFileFromBucketActivity(
{
try
{
await storageHandler.RemoveFileAsync(
await storageService.RemoveFileAsync(
executeContext.Arguments.Bucket,
executeContext.Arguments.Key,
executeContext.CancellationToken);
@@ -7,7 +7,7 @@ using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Files;
internal class TransferFilesToBucketActivity(
IStorageService storageHandler,
IStorageService storageService,
ILogger<TransferFilesToBucketActivity> logger)
: IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog>
{
@@ -16,13 +16,13 @@ internal class TransferFilesToBucketActivity(
public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext)
{
var folderPath = executeContext.GetVariable<string>(executeContext.Arguments.SourceDirectoryVariable);
ArgumentNullException.ThrowIfNull(folderPath, nameof(folderPath));
ArgumentNullException.ThrowIfNull(folderPath);
var destinationKey = executeContext.Arguments.DestinationKey;
ArgumentNullException.ThrowIfNull(destinationKey, nameof(destinationKey));
ArgumentNullException.ThrowIfNull(destinationKey);
try
{
var uploadedKeys = await storageHandler.TransferFilesAsync(
var uploadedKeys = await storageService.TransferFilesAsync(
folderPath,
executeContext.Arguments.DestinationBucket,
destinationKey,
@@ -47,7 +47,7 @@ internal class TransferFilesToBucketActivity(
{
foreach (var uploadedKey in compensateContext.Log.UploadedKeys)
{
await storageHandler.RemoveFileAsync(
await storageService.RemoveFileAsync(
compensateContext.Log.DestinationBucket,
uploadedKey,
compensateContext.CancellationToken);
@@ -6,7 +6,7 @@ using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Files;
internal class UploadFileToBucketActivity(IStorageService storageHandler,
internal class UploadFileToBucketActivity(IStorageService storageService,
ILogger<UploadFileToBucketActivity> logger)
: IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
{
@@ -15,7 +15,7 @@ internal class UploadFileToBucketActivity(IStorageService storageHandler,
public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> executeContext)
{
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
ArgumentNullException.ThrowIfNull(sourceFilePath);
var fileName = Path.GetFileName(sourceFilePath);
var destinationKey = string.Join('/', new[] { executeContext.Arguments.DestinationRoute, fileName }
.Where(static s => !string.IsNullOrWhiteSpace(s))
@@ -23,7 +23,7 @@ internal class UploadFileToBucketActivity(IStorageService storageHandler,
try
{
await storageHandler.UploadFileAsync(
await storageService.UploadFileAsync(
sourceFilePath,
executeContext.Arguments.DestinationBucket,
destinationKey,
@@ -43,7 +43,7 @@ internal class UploadFileToBucketActivity(IStorageService storageHandler,
{
try
{
await storageHandler.RemoveFileAsync(
await storageService.RemoveFileAsync(
compensateContext.Log.DestinationBucket,
compensateContext.Log.DestinationKey,
compensateContext.CancellationToken);
@@ -0,0 +1,97 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.Abstractions;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.LifeCycle;
/// <summary>Sets the life-cycle status of an entity (removing, failed, ...). A missing entity is skipped.</summary>
internal abstract class MarkLifeCycleActivity<TEntity>(IDatabase database, ILogger logger)
: IExecuteActivity<MarkLifeCycleArguments>
where TEntity : class, IHasLifeCycle
{
public async Task<ExecutionResult> Execute(ExecuteContext<MarkLifeCycleArguments> executeContext)
{
var arguments = executeContext.Arguments;
var entity = await database.Set<TEntity>()
.SingleOrDefaultAsync(e => e.Id == arguments.Id, executeContext.CancellationToken);
if (entity == null)
{
logger.LogInformation("{Entity} {Id} not found while marking it as {Status}, skipping", typeof(TEntity).Name, arguments.Id, arguments.Status);
return executeContext.Completed();
}
entity.LifeCycleStatus = arguments.Status;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked {Entity} {Id} as {Status}", typeof(TEntity).Name, arguments.Id, arguments.Status);
return executeContext.Completed();
}
}
/// <summary>Deletes an entity row. A missing entity is skipped.</summary>
internal abstract class DeleteEntityActivity<TEntity>(IDatabase database, ILogger logger)
: IExecuteActivity<DeleteEntityArguments>
where TEntity : class, IHasLifeCycle
{
public async Task<ExecutionResult> Execute(ExecuteContext<DeleteEntityArguments> executeContext)
{
var id = executeContext.Arguments.Id;
var entity = await database.Set<TEntity>()
.SingleOrDefaultAsync(e => e.Id == id, executeContext.CancellationToken);
if (entity == null)
{
logger.LogInformation("{Entity} {Id} not found while deleting it, skipping", typeof(TEntity).Name, id);
return executeContext.Completed();
}
database.Set<TEntity>().Remove(entity);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Deleted {Entity} {Id} from the database", typeof(TEntity).Name, id);
return executeContext.Completed();
}
}
// One concrete activity per entity keeps a separate queue for each of them:
// the endpoint name is the kebab-cased class name.
internal sealed class MarkTrackLifeCycleActivity(IDatabase database, ILogger<MarkTrackLifeCycleActivity> logger)
: MarkLifeCycleActivity<Track>(database, logger)
{
public const string ExecuteEndpointName = "mark-track-life-cycle";
}
internal sealed class MarkPlayListLifeCycleActivity(IDatabase database, ILogger<MarkPlayListLifeCycleActivity> logger)
: MarkLifeCycleActivity<PlayList>(database, logger)
{
public const string ExecuteEndpointName = "mark-play-list-life-cycle";
}
internal sealed class MarkAlbumLifeCycleActivity(IDatabase database, ILogger<MarkAlbumLifeCycleActivity> logger)
: MarkLifeCycleActivity<Album>(database, logger)
{
public const string ExecuteEndpointName = "mark-album-life-cycle";
}
internal sealed class DeleteTrackActivity(IDatabase database, ILogger<DeleteTrackActivity> logger)
: DeleteEntityActivity<Track>(database, logger)
{
public const string ExecuteEndpointName = "delete-track";
}
internal sealed class DeletePlayListActivity(IDatabase database, ILogger<DeletePlayListActivity> logger)
: DeleteEntityActivity<PlayList>(database, logger)
{
public const string ExecuteEndpointName = "delete-play-list";
}
internal sealed class DeleteAlbumActivity(IDatabase database, ILogger<DeleteAlbumActivity> logger)
: DeleteEntityActivity<Album>(database, logger)
{
public const string ExecuteEndpointName = "delete-album";
}
@@ -1,6 +1,7 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
@@ -12,39 +13,27 @@ internal class GeneratePictureWorkflowPathsActivity(
internal const string ResizedPictureFileExtension = ".webp";
public async Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext)
{
try
public Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext)
{
var workingDirectory = Path.Combine(executeContext.Arguments.TemporaryRootDirectory, Guid.NewGuid().ToString());
var sourceFilePath = Path.Combine(
workingDirectory, Guid.NewGuid().ToString() + Path.GetExtension(executeContext.Arguments.SourceKey));
var smallPictureFilePath = Path.Combine(
workingDirectory, Guid.NewGuid().ToString() + ResizedPictureFileExtension);
var mediumPictureFilePath = Path.Combine(
workingDirectory, Guid.NewGuid().ToString() + ResizedPictureFileExtension);
var largePictureFilePath = Path.Combine(
workingDirectory, Guid.NewGuid().ToString() + ResizedPictureFileExtension);
var smallPictureFilePath = Path.Combine(workingDirectory, Guid.NewGuid().ToString() + ResizedPictureFileExtension);
var mediumPictureFilePath = Path.Combine(workingDirectory, Guid.NewGuid().ToString() + ResizedPictureFileExtension);
var largePictureFilePath = Path.Combine(workingDirectory, Guid.NewGuid().ToString() + ResizedPictureFileExtension);
Directory.CreateDirectory(workingDirectory);
logger.LogDebug("Generated picture workflow paths in {WorkingDirectory}",
workingDirectory);
return executeContext.CompletedWithVariables(new Dictionary<string, object>
logger.LogDebug("Generated picture workflow paths in {WorkingDirectory}", workingDirectory);
return Task.FromResult(executeContext.CompletedWithVariables(new Dictionary<string, object>
{
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
[RoutingSlipVariableNames.Picture.OriginalFilePath] = sourceFilePath,
[RoutingSlipVariableNames.Picture.SmallResizedFilePath] = smallPictureFilePath,
[RoutingSlipVariableNames.Picture.MediumResizedFilePath] = mediumPictureFilePath,
[RoutingSlipVariableNames.Picture.LargeResizedFilePath] = largePictureFilePath
});
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to generate picture workflow paths for {SourceKey}",
executeContext.Arguments.SourceKey);
throw;
}
}));
}
}
@@ -16,9 +16,9 @@ internal class ResizePictureActivity(
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
{
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
ArgumentNullException.ThrowIfNull(sourceFilePath);
var destinationFilePath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
ArgumentNullException.ThrowIfNull(destinationFilePath, nameof(destinationFilePath));
ArgumentNullException.ThrowIfNull(destinationFilePath);
try
{
@@ -1,110 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Application.Events;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
internal class UpdateAlbumPictureActivity(
IDatabase database,
IPublishEndpoint publishEndpoint,
ILogger<UpdateAlbumPictureActivity> logger)
: IActivity<UpdateAlbumPictureArguments, UpdateAlbumPictureLog>
{
public const string ExecuteEndpointName = "update-album-picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateAlbumPictureArguments> executeContext)
{
var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable);
ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable));
var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable);
ArgumentNullException.ThrowIfNull(mediumResizedVariable, nameof(mediumResizedVariable));
var largeResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.LargePictureVariable);
ArgumentNullException.ThrowIfNull(largeResizedVariable, nameof(largeResizedVariable));
try
{
var album = await database.Albums.FindAsync(
[executeContext.Arguments.AlbumId],
cancellationToken: executeContext.CancellationToken);
if (album == null)
{
logger.LogWarning("Album {AlbumId} not found",
executeContext.Arguments.AlbumId);
throw new InvalidOperationException($"Album with id {executeContext.Arguments.AlbumId} not found");
}
if (album.Pictures == null)
{
logger.LogWarning("Album {AlbumId} has no pending picture upload",
executeContext.Arguments.AlbumId);
throw new InvalidOperationException($"Album with id {executeContext.Arguments.AlbumId} has no pending picture upload");
}
var log = new UpdateAlbumPictureLog(
album.Id,
album.Pictures.OriginalName,
album.Pictures.SmallName,
album.Pictures.MediumName,
album.Pictures.LargeName);
album.Pictures.OriginalName = Path.GetFileName(executeContext.Arguments.OriginalPictureKey);
album.Pictures.SmallName = Path.GetFileName(smallResizedVariable);
album.Pictures.MediumName = Path.GetFileName(mediumResizedVariable);
album.Pictures.LargeName = Path.GetFileName(largeResizedVariable);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated album {AlbumId} pictures",
executeContext.Arguments.AlbumId);
await publishEndpoint.Publish(
new AlbumPictureProcessed(executeContext.Arguments.AlbumId),
executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to update album {AlbumId} pictures",
executeContext.Arguments.AlbumId);
throw;
}
}
public async Task<CompensationResult> Compensate(CompensateContext<UpdateAlbumPictureLog> compensateContext)
{
try
{
var album = await database.Albums.FindAsync(
[compensateContext.Log.AlbumId],
cancellationToken: compensateContext.CancellationToken);
if (album == null)
{
return compensateContext.Compensated();
}
album.Pictures = new EntityPictures
{
OriginalName = compensateContext.Log.PreviousOriginalPictureKey,
SmallName = compensateContext.Log.PreviousSmallPictureKey,
MediumName = compensateContext.Log.PreviousMediumPictureKey,
LargeName = compensateContext.Log.PreviousLargePictureKey
};
await database.SaveChangesAsync(compensateContext.CancellationToken);
return compensateContext.Compensated();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate album {AlbumId} pictures", compensateContext.Log.AlbumId);
return compensateContext.Failed(exception);
}
}
}
@@ -0,0 +1,160 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Application.Events;
using Musify.Domain.Entities;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
internal sealed record PictureNames(string? Original, string? Small, string? Medium, string? Large);
/// <summary>Stores the processed picture names on the owner and reports it; compensation restores the previous names.</summary>
internal abstract class UpdatePicturesActivity<TEntity>(
IDatabase database,
IPublishEndpoint publishEndpoint,
ILogger logger)
: IActivity<UpdatePicturesArguments, UpdatePicturesLog>
where TEntity : class
{
protected IDatabase Database { get; } = database;
protected abstract Task<TEntity?> FindAsync(Guid id, CancellationToken cancellationToken);
/// <summary>Reads the current names; throws when the entity has no pending picture.</summary>
protected abstract PictureNames Read(TEntity entity);
protected abstract void Write(TEntity entity, PictureNames names, bool succeeded);
protected abstract object ProcessedEvent(Guid id);
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePicturesArguments> executeContext)
{
var arguments = executeContext.Arguments;
var small = RequireVariable(executeContext, arguments.SmallPictureVariable);
var medium = RequireVariable(executeContext, arguments.MediumPictureVariable);
var large = RequireVariable(executeContext, arguments.LargePictureVariable);
var entity = await FindAsync(arguments.SubjectId, executeContext.CancellationToken)
?? throw new InvalidOperationException($"{typeof(TEntity).Name} {arguments.SubjectId} not found");
var previous = Read(entity);
Write(
entity,
new PictureNames(
Path.GetFileName(arguments.OriginalPictureKey),
Path.GetFileName(small),
Path.GetFileName(medium),
Path.GetFileName(large)),
succeeded: true);
await Database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated {Entity} {Id} pictures", typeof(TEntity).Name, arguments.SubjectId);
await publishEndpoint.Publish(ProcessedEvent(arguments.SubjectId), executeContext.CancellationToken);
return executeContext.Completed(new UpdatePicturesLog(
arguments.SubjectId, previous.Original, previous.Small, previous.Medium, previous.Large));
}
public async Task<CompensationResult> Compensate(CompensateContext<UpdatePicturesLog> compensateContext)
{
var log = compensateContext.Log;
try
{
var entity = await FindAsync(log.SubjectId, compensateContext.CancellationToken);
if (entity == null)
return compensateContext.Compensated();
var previous = new PictureNames(log.PreviousOriginalName, log.PreviousSmallName, log.PreviousMediumName, log.PreviousLargeName);
Write(entity, previous, succeeded: false);
await Database.SaveChangesAsync(compensateContext.CancellationToken);
return compensateContext.Compensated();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate {Entity} {Id} pictures", typeof(TEntity).Name, log.SubjectId);
return compensateContext.Failed(exception);
}
}
protected static PictureNames ReadPictures(EntityPictures? pictures) =>
pictures == null
? throw new InvalidOperationException("The entity has no pending picture upload")
: new PictureNames(pictures.OriginalName, pictures.SmallName, pictures.MediumName, pictures.LargeName);
protected static void WritePictures(EntityPictures? pictures, PictureNames names)
{
if (pictures == null)
return;
pictures.OriginalName = names.Original ?? pictures.OriginalName;
pictures.SmallName = names.Small;
pictures.MediumName = names.Medium;
pictures.LargeName = names.Large;
}
private static string RequireVariable(ExecuteContext<UpdatePicturesArguments> executeContext, string variableName) =>
executeContext.GetVariable<string>(variableName)
?? throw new InvalidOperationException($"Missing routing slip variable {variableName}");
}
internal sealed class UpdateTrackPictureActivity(IDatabase database, IPublishEndpoint publishEndpoint, ILogger<UpdateTrackPictureActivity> logger)
: UpdatePicturesActivity<Track>(database, publishEndpoint, logger)
{
public const string ExecuteEndpointName = "update-track-picture";
protected override Task<Track?> FindAsync(Guid id, CancellationToken cancellationToken) =>
Database.Tracks.SingleOrDefaultAsync(track => track.Id == id, cancellationToken);
protected override PictureNames Read(Track track) =>
new(track.Pictures.OriginalName, track.Pictures.SmallName, track.Pictures.MediumName, track.Pictures.LargeName);
protected override void Write(Track track, PictureNames names, bool succeeded)
{
track.Pictures.OriginalName = names.Original;
track.Pictures.SmallName = names.Small;
track.Pictures.MediumName = names.Medium;
track.Pictures.LargeName = names.Large;
track.Pictures.ProcessingStatus = succeeded ? ProcessingStatus.Completed : ProcessingStatus.Failed;
}
protected override object ProcessedEvent(Guid id) => new TrackPictureProcessed(id);
}
internal sealed class UpdatePlayListPictureActivity(IDatabase database, IPublishEndpoint publishEndpoint, ILogger<UpdatePlayListPictureActivity> logger)
: UpdatePicturesActivity<PlayList>(database, publishEndpoint, logger)
{
public const string ExecuteEndpointName = "update-play-list-picture";
protected override Task<PlayList?> FindAsync(Guid id, CancellationToken cancellationToken) =>
Database.PlayLists.SingleOrDefaultAsync(playList => playList.Id == id, cancellationToken);
protected override PictureNames Read(PlayList playList) => ReadPictures(playList.Pictures);
protected override void Write(PlayList playList, PictureNames names, bool succeeded) =>
WritePictures(playList.Pictures, names);
protected override object ProcessedEvent(Guid id) => new PlayListPictureProcessed(id);
}
internal sealed class UpdateAlbumPictureActivity(IDatabase database, IPublishEndpoint publishEndpoint, ILogger<UpdateAlbumPictureActivity> logger)
: UpdatePicturesActivity<Album>(database, publishEndpoint, logger)
{
public const string ExecuteEndpointName = "update-album-picture";
protected override Task<Album?> FindAsync(Guid id, CancellationToken cancellationToken) =>
Database.Albums.SingleOrDefaultAsync(album => album.Id == id, cancellationToken);
protected override PictureNames Read(Album album) => ReadPictures(album.Pictures);
protected override void Write(Album album, PictureNames names, bool succeeded) =>
WritePictures(album.Pictures, names);
protected override object ProcessedEvent(Guid id) => new AlbumPictureProcessed(id);
}
@@ -1,110 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Application.Events;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
internal class UpdatePlayListPictureActivity(
IDatabase database,
IPublishEndpoint publishEndpoint,
ILogger<UpdatePlayListPictureActivity> logger)
: IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog>
{
public const string ExecuteEndpointName = "update-play-list-picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext)
{
var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable);
ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable));
var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable);
ArgumentNullException.ThrowIfNull(mediumResizedVariable, nameof(mediumResizedVariable));
var largeResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.LargePictureVariable);
ArgumentNullException.ThrowIfNull(largeResizedVariable, nameof(largeResizedVariable));
try
{
var playList = await database.PlayLists.FindAsync(
[executeContext.Arguments.PlayListId],
cancellationToken: executeContext.CancellationToken);
if (playList == null)
{
logger.LogWarning("Playlist {PlayListId} not found",
executeContext.Arguments.PlayListId);
throw new InvalidOperationException($"PlayList with id {executeContext.Arguments.PlayListId} not found");
}
if (playList.Pictures == null)
{
logger.LogWarning("Playlist {PlayListId} has no pending picture upload",
executeContext.Arguments.PlayListId);
throw new InvalidOperationException($"PlayList with id {executeContext.Arguments.PlayListId} has no pending picture upload");
}
var log = new UpdatePlayListPictureLog(
playList.Id,
playList.Pictures.OriginalName,
playList.Pictures.SmallName,
playList.Pictures.MediumName,
playList.Pictures.LargeName);
playList.Pictures.OriginalName = Path.GetFileName(executeContext.Arguments.OriginalPictureKey);
playList.Pictures.SmallName = Path.GetFileName(smallResizedVariable);
playList.Pictures.MediumName = Path.GetFileName(mediumResizedVariable);
playList.Pictures.LargeName = Path.GetFileName(largeResizedVariable);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated playlist {PlayListId} pictures",
executeContext.Arguments.PlayListId);
await publishEndpoint.Publish(
new PlayListPictureProcessed(executeContext.Arguments.PlayListId),
executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to update playlist {PlayListId} pictures",
executeContext.Arguments.PlayListId);
throw;
}
}
public async Task<CompensationResult> Compensate(CompensateContext<UpdatePlayListPictureLog> compensateContext)
{
try
{
var playList = await database.PlayLists.FindAsync(
[compensateContext.Log.PlayListId],
cancellationToken: compensateContext.CancellationToken);
if (playList == null)
{
return compensateContext.Compensated();
}
playList.Pictures = new EntityPictures
{
OriginalName = compensateContext.Log.PreviousOriginalPictureKey,
SmallName = compensateContext.Log.PreviousSmallPictureKey,
MediumName = compensateContext.Log.PreviousMediumPictureKey,
LargeName = compensateContext.Log.PreviousLargePictureKey
};
await database.SaveChangesAsync(compensateContext.CancellationToken);
return compensateContext.Compensated();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate playlist {PlayListId} pictures", compensateContext.Log.PlayListId);
return compensateContext.Failed(exception);
}
}
}
@@ -1,100 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Application.Events;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
internal class UpdateTrackPictureActivity(
IDatabase database,
IPublishEndpoint publishEndpoint,
ILogger<UpdateTrackPictureActivity> logger)
: IActivity<UpdateTrackPictureArguments, UpdateTrackPictureLog>
{
public const string ExecuteEndpointName = "update-track-picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateTrackPictureArguments> executeContext)
{
var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable);
ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable));
var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable);
ArgumentNullException.ThrowIfNull(mediumResizedVariable, nameof(mediumResizedVariable));
var largeResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.LargePictureVariable);
ArgumentNullException.ThrowIfNull(largeResizedVariable, nameof(largeResizedVariable));
try
{
var track = await database.Tracks.FindAsync(
[executeContext.Arguments.TrackId],
cancellationToken: executeContext.CancellationToken);
if (track == null)
{
logger.LogWarning("Track {TrackId} not found",
executeContext.Arguments.TrackId);
throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found");
}
var log = new UpdateTrackPictureLog(
track.Id,
track.Pictures.OriginalName,
track.Pictures.SmallName,
track.Pictures.MediumName,
track.Pictures.LargeName);
track.Pictures.OriginalName = Path.GetFileName(executeContext.Arguments.OriginalPictureKey);
track.Pictures.SmallName = Path.GetFileName(smallResizedVariable);
track.Pictures.MediumName = Path.GetFileName(mediumResizedVariable);
track.Pictures.LargeName = Path.GetFileName(largeResizedVariable);
track.Pictures.ProcessingStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated track {TrackId} pictures",
executeContext.Arguments.TrackId);
await publishEndpoint.Publish(
new TrackPictureProcessed(executeContext.Arguments.TrackId),
executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to update track {TrackId} pictures",
executeContext.Arguments.TrackId);
throw;
}
}
public async Task<CompensationResult> Compensate(CompensateContext<UpdateTrackPictureLog> compensateContext)
{
try
{
var track = await database.Tracks.FindAsync(
[compensateContext.Log.TrackId],
cancellationToken: compensateContext.CancellationToken);
if (track == null)
return compensateContext.Compensated();
track.Pictures.OriginalName = compensateContext.Log.PreviousOriginalPictureKey;
track.Pictures.SmallName = compensateContext.Log.PreviousSmallPictureKey;
track.Pictures.MediumName = compensateContext.Log.PreviousMediumPictureKey;
track.Pictures.LargeName = compensateContext.Log.PreviousLargePictureKey;
track.Pictures.ProcessingStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(compensateContext.CancellationToken);
return compensateContext.Compensated();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate track {TrackId} pictures", compensateContext.Log.TrackId);
return compensateContext.Failed(exception);
}
}
}
@@ -1,35 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.PlayLists;
internal class DeletePlayListFromDbActivity(
IDatabase database,
ILogger<DeletePlayListFromDbActivity> logger)
: IExecuteActivity<DeletePlayListFromDbArguments>
{
public const string ExecuteEndpointName = "delete-play-list-from-db";
public async Task<ExecutionResult> Execute(ExecuteContext<DeletePlayListFromDbArguments> executeContext)
{
var playList = await database.PlayLists
.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.PlayListId, executeContext.CancellationToken);
if (playList == null)
{
logger.LogInformation(
"PlayList {PlayListId} not found while deleting from DB, skipping",
executeContext.Arguments.PlayListId);
return executeContext.Completed();
}
database.PlayLists.Remove(playList);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Deleted playlist {PlayListId} from DB", playList.Id);
return executeContext.Completed();
}
}
@@ -1,31 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.PlayLists;
internal class MarkPlayListAsFailedActivity(
IDatabase database,
ILogger<MarkPlayListAsFailedActivity> logger)
: IExecuteActivity<MarkPlayListAsFailedArguments>
{
public const string ExecuteEndpointName = "mark-play-list-as-failed";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkPlayListAsFailedArguments> executeContext)
{
var playList = await database.PlayLists
.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.PlayListId, executeContext.CancellationToken);
if (playList != null)
{
playList.LifeCycleStatus = LifeCycleStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked playlist {PlayListId} as failed", executeContext.Arguments.PlayListId);
}
return executeContext.Completed();
}
}
@@ -1,36 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.PlayLists;
internal class MarkPlayListAsRemovingActivity(
IDatabase database,
ILogger<MarkPlayListAsRemovingActivity> logger)
: IExecuteActivity<MarkPlayListAsRemovingArguments>
{
public const string ExecuteEndpointName = "mark-play-list-as-removing";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkPlayListAsRemovingArguments> executeContext)
{
var playList = await database.PlayLists
.SingleOrDefaultAsync(p => p.Id == executeContext.Arguments.PlayListId, executeContext.CancellationToken);
if (playList == null)
{
logger.LogInformation(
"PlayList {PlayListId} not found while marking as removing, skipping",
executeContext.Arguments.PlayListId);
return executeContext.Completed();
}
playList.LifeCycleStatus = LifeCycleStatus.Removing;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked playlist {PlayListId} as removing", playList.Id);
return executeContext.Completed();
}
}
@@ -1,35 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Tracks;
internal class DeleteTrackFromDbActivity(
IDatabase database,
ILogger<DeleteTrackFromDbActivity> logger)
: IExecuteActivity<DeleteTrackFromDbArguments>
{
public const string ExecuteEndpointName = "delete-track-from-db";
public async Task<ExecutionResult> Execute(ExecuteContext<DeleteTrackFromDbArguments> executeContext)
{
var track = await database.Tracks
.SingleOrDefaultAsync(t => t.Id == executeContext.Arguments.TrackId, executeContext.CancellationToken);
if (track == null)
{
logger.LogInformation(
"Track {TrackId} not found while deleting from DB, skipping",
executeContext.Arguments.TrackId);
return executeContext.Completed();
}
database.Tracks.Remove(track);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Deleted track {TrackId} from DB", track.Id);
return executeContext.Completed();
}
}
@@ -1,31 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Tracks;
internal class MarkTrackAsFailedActivity(
IDatabase database,
ILogger<MarkTrackAsFailedActivity> logger)
: IExecuteActivity<MarkTrackAsFailedArguments>
{
public const string ExecuteEndpointName = "mark-track-as-failed";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkTrackAsFailedArguments> executeContext)
{
var track = await database.Tracks
.SingleOrDefaultAsync(t => t.Id == executeContext.Arguments.TrackId, executeContext.CancellationToken);
if (track != null)
{
track.LifeCycleStatus = LifeCycleStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked track {TrackId} as failed", executeContext.Arguments.TrackId);
}
return executeContext.Completed();
}
}
@@ -1,36 +0,0 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Tracks;
internal class MarkTrackAsRemovingActivity(
IDatabase database,
ILogger<MarkTrackAsRemovingActivity> logger)
: IExecuteActivity<MarkTrackAsRemovingArguments>
{
public const string ExecuteEndpointName = "mark-track-as-removing";
public async Task<ExecutionResult> Execute(ExecuteContext<MarkTrackAsRemovingArguments> executeContext)
{
var track = await database.Tracks
.SingleOrDefaultAsync(t => t.Id == executeContext.Arguments.TrackId, executeContext.CancellationToken);
if (track == null)
{
logger.LogInformation(
"Track {TrackId} not found while marking as removing, skipping",
executeContext.Arguments.TrackId);
return executeContext.Completed();
}
track.LifeCycleStatus = LifeCycleStatus.Removing;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Marked track {TrackId} as removing", track.Id);
return executeContext.Completed();
}
}
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record DeleteAlbumFromDbArguments(
Guid AlbumId);
@@ -1,3 +1,3 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
public record MarkTrackAsFailedArguments(Guid TrackId);
internal record DeleteEntityArguments(Guid Id);
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record DeletePlayListFromDbArguments(
Guid PlayListId);
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record DeleteTrackFromDbArguments(
Guid TrackId);
@@ -1,3 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
public record MarkAlbumAsFailedArguments(Guid AlbumId);
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record MarkAlbumAsRemovingArguments(
Guid AlbumId);
@@ -0,0 +1,5 @@
using Musify.Domain.ValueObjects;
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record MarkLifeCycleArguments(Guid Id, LifeCycleStatus Status);
@@ -1,3 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
public record MarkPlayListAsFailedArguments(Guid PlayListId);
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record MarkPlayListAsRemovingArguments(
Guid PlayListId);
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
internal record MarkTrackAsRemovingArguments(
Guid TrackId);
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
public record UpdateAlbumPictureArguments(
Guid AlbumId,
string OriginalPictureKey,
string SmallPictureVariable,
string MediumPictureVariable,
string LargePictureVariable);
@@ -1,7 +1,7 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
public record UpdateTrackPictureArguments(
Guid TrackId,
public record UpdatePicturesArguments(
Guid SubjectId,
string OriginalPictureKey,
string SmallPictureVariable,
string MediumPictureVariable,
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Arguments;
public record UpdatePlayListPictureArguments(
Guid PlayListId,
string OriginalPictureKey,
string SmallPictureVariable,
string MediumPictureVariable,
string LargePictureVariable);
@@ -1,37 +1,18 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Infrastructure.MassTransit.Activities.Albums;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Activities.LifeCycle;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class AlbumProcessingFailedConsumer(IBus bus) : IConsumer<AlbumProcessingFailed>
{
public async Task Consume(ConsumeContext<AlbumProcessingFailed> context)
public Task Consume(ConsumeContext<AlbumProcessingFailed> context)
{
var message = context.Message;
var builder = new RoutingSlipBuilder(NewId.NextGuid());
builder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
if (!string.IsNullOrWhiteSpace(message.Bucket) && !string.IsNullOrWhiteSpace(message.PictureKey))
{
builder.AddActivity(
ActivityNames.RemoveAlbumOriginalPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(message.Bucket, message.PictureKey));
}
builder.AddActivity(
ActivityNames.MarkAlbumAsFailed,
EndpointHelper.BuildExecuteActivityUri(MarkAlbumAsFailedActivity.ExecuteEndpointName),
new MarkAlbumAsFailedArguments(message.AlbumId));
await bus.Execute(builder.Build());
return ProcessingFailedCleanup.ExecuteAsync(
bus, message.AlbumId, ActivityNames.MarkAlbumAsFailed, MarkAlbumLifeCycleActivity.ExecuteEndpointName, message.Bucket,
(ActivityNames.RemoveAlbumOriginalPicture, message.PictureKey));
}
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class CreateAlbumConsumer(
IBus bus,
AlbumPictureSourceRoutingSlipBuilder routingSlipBuilder)
public class CreateAlbumConsumer(IBus bus, PictureSourceRoutingSlipBuilder routingSlipBuilder)
: IConsumer<CreateAlbumResourcesEvent>
{
public async Task Consume(ConsumeContext<CreateAlbumResourcesEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<CreateAlbumResourcesEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class CreatePlayListConsumer(
IBus bus,
PlayListPictureSourceRoutingSlipBuilder routingSlipBuilder)
public class CreatePlayListConsumer(IBus bus, PictureSourceRoutingSlipBuilder routingSlipBuilder)
: IConsumer<CreatePlayListResourcesEvent>
{
public async Task Consume(ConsumeContext<CreatePlayListResourcesEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<CreatePlayListResourcesEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class CreateTrackConsumer(
IBus bus,
CreateTrackRoutingSlipBuilder routingSlipBuilder)
public class CreateTrackConsumer(IBus bus, CreateTrackRoutingSlipBuilder routingSlipBuilder)
: IConsumer<CreateTrackResourcesEvent>
{
public async Task Consume(ConsumeContext<CreateTrackResourcesEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<CreateTrackResourcesEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,21 +4,15 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class DeleteAlbumConsumer(
IBus bus,
DeleteAlbumRoutingSlipBuilder routingSlipBuilder)
public class DeleteAlbumConsumer(IBus bus, DeleteRoutingSlipBuilder routingSlipBuilder)
: IConsumer<DeleteAlbumEvent>
{
public async Task Consume(ConsumeContext<DeleteAlbumEvent> consumeContext)
public async Task Consume(ConsumeContext<DeleteAlbumEvent> context)
{
var routingSlip = (await routingSlipBuilder
.BuildAsync(
consumeContext.Message.AlbumId,
consumeContext.Message.UserId,
consumeContext.CorrelationId,
consumeContext.CancellationToken))
.Build();
var message = context.Message;
var routingSlip = await routingSlipBuilder.BuildAlbumAsync(
message.AlbumId, message.UserId, context.CorrelationId, context.CancellationToken);
await bus.Execute(routingSlip);
await bus.Execute(routingSlip.Build());
}
}
@@ -4,21 +4,15 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class DeletePlayListConsumer(
IBus bus,
DeletePlayListRoutingSlipBuilder routingSlipBuilder)
public class DeletePlayListConsumer(IBus bus, DeleteRoutingSlipBuilder routingSlipBuilder)
: IConsumer<DeletePlayListEvent>
{
public async Task Consume(ConsumeContext<DeletePlayListEvent> consumeContext)
public async Task Consume(ConsumeContext<DeletePlayListEvent> context)
{
var routingSlip = (await routingSlipBuilder
.BuildAsync(
consumeContext.Message.PlayListId,
consumeContext.Message.UserId,
consumeContext.CorrelationId,
consumeContext.CancellationToken))
.Build();
var message = context.Message;
var routingSlip = await routingSlipBuilder.BuildPlayListAsync(
message.PlayListId, message.UserId, context.CorrelationId, context.CancellationToken);
await bus.Execute(routingSlip);
await bus.Execute(routingSlip.Build());
}
}
@@ -4,17 +4,15 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class DeleteTrackConsumer(
IBus bus,
DeleteTrackRoutingSlipBuilder routingSlipBuilder)
public class DeleteTrackConsumer(IBus bus, DeleteRoutingSlipBuilder routingSlipBuilder)
: IConsumer<DeleteTrackEvent>
{
public async Task Consume(ConsumeContext<DeleteTrackEvent> consumeContext)
public async Task Consume(ConsumeContext<DeleteTrackEvent> context)
{
var routingSlip = (await routingSlipBuilder
.BuildAsync(consumeContext.Message.TrackId, consumeContext.Message.UserId, consumeContext.CorrelationId, consumeContext.CancellationToken))
.Build();
var message = context.Message;
var routingSlip = await routingSlipBuilder.BuildTrackAsync(
message.TrackId, message.UserId, context.CorrelationId, context.CancellationToken);
await bus.Execute(routingSlip);
await bus.Execute(routingSlip.Build());
}
}
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Consumers;
internal static class EndpointHelper
{
internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{endpointName}_execute");
internal static Uri BuildConsumerUri(string endpointName) => new($"queue:{endpointName}");
}
@@ -1,37 +1,18 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.PlayLists;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Activities.LifeCycle;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class PlayListProcessingFailedConsumer(IBus bus) : IConsumer<PlayListProcessingFailed>
{
public async Task Consume(ConsumeContext<PlayListProcessingFailed> context)
public Task Consume(ConsumeContext<PlayListProcessingFailed> context)
{
var message = context.Message;
var builder = new RoutingSlipBuilder(NewId.NextGuid());
builder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
if (!string.IsNullOrWhiteSpace(message.Bucket) && !string.IsNullOrWhiteSpace(message.PictureKey))
{
builder.AddActivity(
ActivityNames.RemovePlayListOriginalPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(message.Bucket, message.PictureKey));
}
builder.AddActivity(
ActivityNames.MarkPlayListAsFailed,
EndpointHelper.BuildExecuteActivityUri(MarkPlayListAsFailedActivity.ExecuteEndpointName),
new MarkPlayListAsFailedArguments(message.PlayListId));
await bus.Execute(builder.Build());
return ProcessingFailedCleanup.ExecuteAsync(
bus, message.PlayListId, ActivityNames.MarkPlayListAsFailed, MarkPlayListLifeCycleActivity.ExecuteEndpointName, message.Bucket,
(ActivityNames.RemovePlayListOriginalPicture, message.PictureKey));
}
}
@@ -0,0 +1,34 @@
using MassTransit;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Consumers;
internal static class ProcessingFailedCleanup
{
/// <summary>Removes whatever files were left behind and marks the entity as failed.</summary>
public static Task ExecuteAsync(
IBus bus,
Guid subjectId,
string markStepName,
string markEndpointName,
string? bucket,
params (string StepName, string? Key)[] files)
{
var builder = RoutingSlips.Create(correlationId: null);
if (!string.IsNullOrWhiteSpace(bucket))
{
foreach (var (stepName, key) in files)
{
if (!string.IsNullOrWhiteSpace(key))
builder.AddRemoveFile(stepName, bucket, key);
}
}
builder.AddStep(markStepName, markEndpointName, new MarkLifeCycleArguments(subjectId, LifeCycleStatus.Failed));
return bus.Execute(builder.Build());
}
}
@@ -1,6 +1,7 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Consumers;
@@ -1,6 +1,7 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Microsoft.Extensions.Logging;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Consumers;
@@ -32,14 +33,9 @@ public class RoutingSlipCleanUpConsumer(
Directory.Delete(pathString, recursive: true);
logger.LogInformation("Deleted temporal directory at path {Path}", pathString);
}
catch (DirectoryNotFoundException directoryNotFoundException)
catch (DirectoryNotFoundException)
{
logger.LogWarning(directoryNotFoundException, "Temporal directory at path {Path} was not found for deletion", pathString);
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to delete temporal directory at path {Path}", pathString);
throw;
logger.LogWarning("Temporal directory at path {Path} was not found for deletion", pathString);
}
return Task.CompletedTask;
@@ -1,45 +1,19 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.Tracks;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Activities.LifeCycle;
using Musify.Infrastructure.MassTransit.RoutingSlip;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class TrackProcessingFailedConsumer(IBus bus) : IConsumer<TrackProcessingFailed>
{
public async Task Consume(ConsumeContext<TrackProcessingFailed> context)
public Task Consume(ConsumeContext<TrackProcessingFailed> context)
{
var message = context.Message;
var builder = new RoutingSlipBuilder(NewId.NextGuid());
builder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
if (!string.IsNullOrWhiteSpace(message.Bucket) && !string.IsNullOrWhiteSpace(message.PictureKey))
{
builder.AddActivity(
ActivityNames.RemoveTrackOriginalPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(message.Bucket, message.PictureKey));
}
if (!string.IsNullOrWhiteSpace(message.Bucket) && !string.IsNullOrWhiteSpace(message.AudioKey))
{
builder.AddActivity(
ActivityNames.RemoveTrackOriginalAudio,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(message.Bucket, message.AudioKey));
}
builder.AddActivity(
ActivityNames.MarkTrackAsFailed,
EndpointHelper.BuildExecuteActivityUri(MarkTrackAsFailedActivity.ExecuteEndpointName),
new MarkTrackAsFailedArguments(message.TrackId));
await bus.Execute(builder.Build());
return ProcessingFailedCleanup.ExecuteAsync(
bus, message.TrackId, ActivityNames.MarkTrackAsFailed, MarkTrackLifeCycleActivity.ExecuteEndpointName, message.Bucket,
(ActivityNames.RemoveTrackOriginalPicture, message.PictureKey),
(ActivityNames.RemoveTrackOriginalAudio, message.AudioKey));
}
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class UpdateAlbumPictureConsumer(
IBus bus,
PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
public class UpdateAlbumPictureConsumer(IBus bus, PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdateAlbumPictureEvent>
{
public async Task Consume(ConsumeContext<UpdateAlbumPictureEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<UpdateAlbumPictureEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class UpdateAlbumPictureSourceConsumer(
IBus bus,
AlbumPictureSourceRoutingSlipBuilder routingSlipBuilder)
public class UpdateAlbumPictureSourceConsumer(IBus bus, PictureSourceRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdateAlbumPictureSourceEvent>
{
public async Task Consume(ConsumeContext<UpdateAlbumPictureSourceEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<UpdateAlbumPictureSourceEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class UpdatePlayListPictureConsumer(
IBus bus,
PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
public class UpdatePlayListPictureConsumer(IBus bus, PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdatePlayListPictureEvent>
{
public async Task Consume(ConsumeContext<UpdatePlayListPictureEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<UpdatePlayListPictureEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class UpdatePlayListPictureSourceConsumer(
IBus bus,
PlayListPictureSourceRoutingSlipBuilder routingSlipBuilder)
public class UpdatePlayListPictureSourceConsumer(IBus bus, PictureSourceRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdatePlayListPictureSourceEvent>
{
public async Task Consume(ConsumeContext<UpdatePlayListPictureSourceEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<UpdatePlayListPictureSourceEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class UpdateTrackAudioConsumer(
IBus bus,
AudioWorkflowRoutingSlipBuilder routingSlipBuilder)
public class UpdateTrackAudioConsumer(IBus bus, AudioWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdateTrackAudioEvent>
{
public async Task Consume(ConsumeContext<UpdateTrackAudioEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<UpdateTrackAudioEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -4,17 +4,9 @@ using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
namespace Musify.Infrastructure.MassTransit.Consumers;
public class UpdateTrackPictureConsumer(
IBus bus,
PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
public class UpdateTrackPictureConsumer(IBus bus, PictureWorkflowRoutingSlipBuilder routingSlipBuilder)
: IConsumer<UpdateTrackPictureEvent>
{
public async Task Consume(ConsumeContext<UpdateTrackPictureEvent> consumeContext)
{
var routingSlip = routingSlipBuilder
.Build(consumeContext.Message, consumeContext.CorrelationId)
.Build();
await bus.Execute(routingSlip);
}
public Task Consume(ConsumeContext<UpdateTrackPictureEvent> context) =>
bus.Execute(routingSlipBuilder.Build(context.Message, context.CorrelationId).Build());
}
@@ -11,6 +11,7 @@ using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
using Musify.Infrastructure.MassTransit.Logs;
using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
using Musify.Infrastructure.MassTransit.Activities.LifeCycle;
namespace Musify.Infrastructure.MassTransit;
@@ -19,61 +20,55 @@ public static partial class MassTransitDependencyInjection
private static void RegisterRoutingSlipBuilders(IServiceCollection services)
{
services.AddScoped<AudioWorkflowRoutingSlipBuilder>();
services.AddScoped<PictureWorkflowRoutingSlipBuilder>();
services.AddScoped<DeleteTrackRoutingSlipBuilder>();
services.AddScoped<DeletePlayListRoutingSlipBuilder>();
services.AddScoped<DeleteAlbumRoutingSlipBuilder>();
services.AddScoped<CreateTrackRoutingSlipBuilder>();
services.AddScoped<PlayListPictureSourceRoutingSlipBuilder>();
services.AddScoped<AlbumPictureSourceRoutingSlipBuilder>();
services.AddScoped<DeleteRoutingSlipBuilder>();
services.AddScoped<PictureSourceRoutingSlipBuilder>();
services.AddScoped<PictureWorkflowRoutingSlipBuilder>();
}
private static void RegisterConsumersAndActivities(IBusRegistrationConfigurator options)
{
options.AddConsumer<UpdatePlayListPictureConsumer>();
options.AddConsumer<UpdateTrackPictureConsumer>();
options.AddConsumer<RoutingSlipCleanUpConsumer>();
options.AddConsumer<UpdateTrackAudioConsumer>();
options.AddConsumer<DeleteTrackConsumer>();
options.AddConsumer<DeletePlayListConsumer>();
options.AddConsumer<DeleteAlbumConsumer>();
options.AddConsumer<CreateTrackConsumer>();
options.AddConsumer<CreatePlayListConsumer>();
options.AddConsumer<UpdatePlayListPictureSourceConsumer>();
options.AddConsumer<ProcessingSlipFaultConsumer>();
options.AddConsumer<TrackProcessingFailedConsumer>();
options.AddConsumer<PlayListProcessingFailedConsumer>();
options.AddConsumer<CreateAlbumConsumer>();
options.AddConsumer<UpdateAlbumPictureSourceConsumer>();
options.AddConsumer<UpdateAlbumPictureConsumer>();
options.AddConsumer<AlbumProcessingFailedConsumer>();
options.AddConsumer<CreateAlbumConsumer>();
options.AddConsumer<CreatePlayListConsumer>();
options.AddConsumer<CreateTrackConsumer>();
options.AddConsumer<DeleteAlbumConsumer>();
options.AddConsumer<DeletePlayListConsumer>();
options.AddConsumer<DeleteTrackConsumer>();
options.AddConsumer<PlayListProcessingFailedConsumer>();
options.AddConsumer<ProcessingSlipFaultConsumer>();
options.AddConsumer<RoutingSlipCleanUpConsumer>();
options.AddConsumer<TrackProcessingFailedConsumer>();
options.AddConsumer<UpdateAlbumPictureConsumer>();
options.AddConsumer<UpdateAlbumPictureSourceConsumer>();
options.AddConsumer<UpdatePlayListPictureConsumer>();
options.AddConsumer<UpdatePlayListPictureSourceConsumer>();
options.AddConsumer<UpdateTrackAudioConsumer>();
options.AddConsumer<UpdateTrackPictureConsumer>();
options.AddExecuteActivity<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>();
options.AddExecuteActivity<DeleteAlbumActivity, DeleteEntityArguments>();
options.AddExecuteActivity<DeletePlayListActivity, DeleteEntityArguments>();
options.AddExecuteActivity<DeleteTrackActivity, DeleteEntityArguments>();
options.AddExecuteActivity<GenerateAudioWorkflowPathsActivity, GenerateAudioWorkflowPathsArguments>();
options.AddExecuteActivity<GeneratePictureWorkflowPathsActivity, GeneratePictureWorkflowPathsArguments>();
options.AddExecuteActivity<MarkTrackAsRemovingActivity, MarkTrackAsRemovingArguments>();
options.AddExecuteActivity<DeleteTrackFromDbActivity, DeleteTrackFromDbArguments>();
options.AddExecuteActivity<MarkPlayListAsRemovingActivity, MarkPlayListAsRemovingArguments>();
options.AddExecuteActivity<DeletePlayListFromDbActivity, DeletePlayListFromDbArguments>();
options.AddExecuteActivity<PublishTrackProcessingEventsActivity, PublishTrackProcessingEventsArguments>();
options.AddExecuteActivity<PublishPlayListPictureProcessingEventActivity, PublishPlayListPictureProcessingEventArguments>();
options.AddExecuteActivity<MarkTrackAsFailedActivity, MarkTrackAsFailedArguments>();
options.AddExecuteActivity<MarkPlayListAsFailedActivity, MarkPlayListAsFailedArguments>();
options.AddExecuteActivity<MarkAlbumAsFailedActivity, MarkAlbumAsFailedArguments>();
options.AddExecuteActivity<MarkAlbumAsRemovingActivity, MarkAlbumAsRemovingArguments>();
options.AddExecuteActivity<DeleteAlbumFromDbActivity, DeleteAlbumFromDbArguments>();
options.AddExecuteActivity<MarkAlbumLifeCycleActivity, MarkLifeCycleArguments>();
options.AddExecuteActivity<MarkPlayListLifeCycleActivity, MarkLifeCycleArguments>();
options.AddExecuteActivity<MarkTrackLifeCycleActivity, MarkLifeCycleArguments>();
options.AddExecuteActivity<PublishAlbumPictureProcessingEventActivity, PublishAlbumPictureProcessingEventArguments>();
options.AddExecuteActivity<PublishPlayListPictureProcessingEventActivity, PublishPlayListPictureProcessingEventArguments>();
options.AddExecuteActivity<PublishTrackProcessingEventsActivity, PublishTrackProcessingEventsArguments>();
options.AddExecuteActivity<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>();
options.AddActivity<ResizePictureActivity, ResizePictureLocalArguments, ResizePictureLog>();
options.AddActivity<UpdatePlayListPictureActivity, UpdatePlayListPictureArguments, UpdatePlayListPictureLog>();
options.AddActivity<UpdateAlbumPictureActivity, UpdateAlbumPictureArguments, UpdateAlbumPictureLog>();
options.AddActivity<UpdateTrackPictureActivity, UpdateTrackPictureArguments, UpdateTrackPictureLog>();
options.AddActivity<ConsumeUploadIntentsActivity, ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>();
options.AddActivity<CopyFileInBucketActivity, CopyFileInBucketArguments, CopyFileInBucketLog>();
options.AddActivity<DownloadFileFromBucketActivity, DownloadFileFromBucketArguments, DownloadFileFromBucketLog>();
options.AddActivity<ResizePictureActivity, ResizePictureLocalArguments, ResizePictureLog>();
options.AddActivity<TranscodeAudioActivity, TranscodeAudioArguments, TranscodeAudioLog>();
options.AddActivity<TransferFilesToBucketActivity, TransferFilesToBucketArguments, TransferFilesToBucketLog>();
options.AddActivity<UpdateAlbumPictureActivity, UpdatePicturesArguments, UpdatePicturesLog>();
options.AddActivity<UpdatePlayListPictureActivity, UpdatePicturesArguments, UpdatePicturesLog>();
options.AddActivity<UpdateTrackAudioActivity, UpdateTrackAudioArguments, UpdateTrackAudioLog>();
options.AddActivity<UpdateTrackPictureActivity, UpdatePicturesArguments, UpdatePicturesLog>();
options.AddActivity<UploadFileToBucketActivity, UploadFileToBucketArguments, UploadFileToBucketLog>();
options.AddActivity<CopyFileInBucketActivity, CopyFileInBucketArguments, CopyFileInBucketLog>();
options.AddActivity<ConsumeUploadIntentsActivity, ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>();
}
}
@@ -18,14 +18,7 @@ public static partial class MassTransitDependencyInjection
services.AddScoped<IEventBus, MassTransitEventBus>();
services.AddMassTransit(options =>
{
options.AddEntityFrameworkOutbox<Database>(outbox =>
{
outbox.UsePostgres();
outbox.UseBusOutbox();
outbox.QueryDelay = TimeSpan.FromSeconds(1);
outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
});
AddOutbox(options);
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
ConfigureRabbitMqHost(
@@ -46,42 +39,20 @@ public static partial class MassTransitDependencyInjection
services.AddMassTransit(options =>
{
options.AddEntityFrameworkOutbox<Database>(outbox =>
{
outbox.UsePostgres();
outbox.UseBusOutbox();
outbox.QueryDelay = TimeSpan.FromSeconds(1);
outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
});
AddOutbox(options);
options.SetKebabCaseEndpointNameFormatter();
RegisterConsumersAndActivities(options);
options.AddSagaStateMachine<TrackProcessingStateMachine, TrackProcessingState>()
.EntityFrameworkRepository(repository =>
{
repository.ConcurrencyMode = ConcurrencyMode.Pessimistic;
repository.ExistingDbContext<Database>();
repository.UsePostgres();
});
.EntityFrameworkRepository(ConfigureSagaRepository<TrackProcessingState>);
options.AddSagaStateMachine<PlayListProcessingStateMachine, PlayListProcessingState>()
.EntityFrameworkRepository(repository =>
{
repository.ConcurrencyMode = ConcurrencyMode.Pessimistic;
repository.ExistingDbContext<Database>();
repository.UsePostgres();
});
.EntityFrameworkRepository(ConfigureSagaRepository<PlayListProcessingState>);
options.AddSagaStateMachine<AlbumProcessingStateMachine, AlbumProcessingState>()
.EntityFrameworkRepository(repository =>
{
repository.ConcurrencyMode = ConcurrencyMode.Pessimistic;
repository.ExistingDbContext<Database>();
repository.UsePostgres();
});
.EntityFrameworkRepository(ConfigureSagaRepository<AlbumProcessingState>);
options.AddConfigureEndpointsCallback((registrationContext, _, endpointConfigurator) =>
{
@@ -102,6 +73,24 @@ public static partial class MassTransitDependencyInjection
return services;
}
private static void AddOutbox(IBusRegistrationConfigurator options) =>
options.AddEntityFrameworkOutbox<Database>(outbox =>
{
outbox.UsePostgres();
outbox.UseBusOutbox();
outbox.QueryDelay = TimeSpan.FromSeconds(1);
outbox.DuplicateDetectionWindow = TimeSpan.FromMinutes(30);
});
private static void ConfigureSagaRepository<TSaga>(IEntityFrameworkSagaRepositoryConfigurator<TSaga> repository)
where TSaga : class, ISaga
{
repository.ConcurrencyMode = ConcurrencyMode.Pessimistic;
repository.ExistingDbContext<Database>();
repository.UsePostgres();
}
private static void UseStandardRetry(IReceiveEndpointConfigurator endpointConfigurator) =>
endpointConfigurator.UseMessageRetry(retryConfigurator =>
retryConfigurator.Exponential(
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Logs;
public record UpdateAlbumPictureLog(
Guid AlbumId,
string PreviousOriginalPictureKey,
string? PreviousSmallPictureKey,
string? PreviousMediumPictureKey,
string? PreviousLargePictureKey);
@@ -0,0 +1,8 @@
namespace Musify.Infrastructure.MassTransit.Logs;
public record UpdatePicturesLog(
Guid SubjectId,
string? PreviousOriginalName,
string? PreviousSmallName,
string? PreviousMediumName,
string? PreviousLargeName);
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Logs;
public record UpdatePlayListPictureLog(
Guid PlayListId,
string PreviousOriginalPictureKey,
string? PreviousSmallPictureKey,
string? PreviousMediumPictureKey,
string? PreviousLargePictureKey);
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.MassTransit.Logs;
public record UpdateTrackPictureLog(
Guid TrackId,
string? PreviousOriginalPictureKey,
string? PreviousSmallPictureKey,
string? PreviousMediumPictureKey,
string? PreviousLargePictureKey);
@@ -1,87 +0,0 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Application.Shared;
using Musify.Infrastructure.MassTransit.Activities.Albums;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.UploadIntents;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
public class AlbumPictureSourceRoutingSlipBuilder
{
public RoutingSlipBuilder Build(CreateAlbumResourcesEvent message, Guid? correlationId)
=> BuildSlip(
message.AlbumId,
message.PictureIntentId,
message.Bucket,
message.PictureSourceKey,
message.PictureDestinationKey,
message.Sizes,
correlationId,
RoutingSlipVariableNames.ProcessKinds.AlbumCreation);
public RoutingSlipBuilder Build(UpdateAlbumPictureSourceEvent message, Guid? correlationId)
=> BuildSlip(
message.AlbumId,
message.PictureIntentId,
message.Bucket,
message.PictureSourceKey,
message.PictureDestinationKey,
message.Sizes,
correlationId,
RoutingSlipVariableNames.ProcessKinds.AlbumPicture);
private static RoutingSlipBuilder BuildSlip(
Guid albumId,
Guid pictureIntentId,
string bucket,
string pictureSourceKey,
string pictureDestinationKey,
ImageSizes sizes,
Guid? correlationId,
string processKind)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
routingSlipBuilder.AddActivity(
ActivityNames.CopyPictureToFinal,
EndpointHelper.BuildExecuteActivityUri(CopyFileInBucketActivity.ExecuteEndpointName),
new CopyFileInBucketArguments(
bucket,
pictureSourceKey,
bucket,
pictureDestinationKey));
routingSlipBuilder.AddActivity(
ActivityNames.ConsumeUploadIntents,
EndpointHelper.BuildExecuteActivityUri(ConsumeUploadIntentsActivity.ExecuteEndpointName),
new ConsumeUploadIntentsArguments([pictureIntentId]));
routingSlipBuilder.AddActivity(
ActivityNames.PublishAlbumPictureProcessingEvent,
EndpointHelper.BuildExecuteActivityUri(PublishAlbumPictureProcessingEventActivity.ExecuteEndpointName),
new PublishAlbumPictureProcessingEventArguments(
albumId,
bucket,
pictureDestinationKey,
sizes));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, albumId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder;
}
}
@@ -1,71 +1,44 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.MassTransit.Activities.Audio;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
public class AudioWorkflowRoutingSlipBuilder(WorkerConfiguration workerConfiguration)
{
public RoutingSlipBuilder Build(UpdateTrackAudioEvent message, Guid? correlationId)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
routingSlipBuilder.AddActivity(
public RoutingSlipBuilder Build(UpdateTrackAudioEvent message, Guid? correlationId) =>
RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.GenerateAudioWorkflowPaths,
EndpointHelper.BuildExecuteActivityUri(GenerateAudioWorkflowPathsActivity.ExecuteEndpointName),
GenerateAudioWorkflowPathsActivity.ExecuteEndpointName,
new GenerateAudioWorkflowPathsArguments(
message.TrackId,
workerConfiguration.Routes.TemporaryFilesDirectory,
message.SourceKey));
routingSlipBuilder.AddActivity(
message.TrackId, workerConfiguration.Routes.TemporaryFilesDirectory, message.SourceKey))
.AddStep(
ActivityNames.DownloadFile,
EndpointHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
DownloadFileFromBucketActivity.ExecuteEndpointName,
new DownloadFileFromBucketArguments(
message.SourceBucket,
message.SourceKey,
RoutingSlipVariableNames.Audio.SourceFilePath));
routingSlipBuilder.AddActivity(
message.SourceBucket, message.SourceKey, RoutingSlipVariableNames.Audio.SourceFilePath))
.AddStep(
ActivityNames.TranscodeAudio,
EndpointHelper.BuildExecuteActivityUri(TranscodeAudioActivity.ExecuteEndpointName),
TranscodeAudioActivity.ExecuteEndpointName,
new TranscodeAudioArguments(
message.TrackId,
RoutingSlipVariableNames.Audio.SourceFilePath,
RoutingSlipVariableNames.Workflow.TemporalDirectory));
routingSlipBuilder.AddActivity(
RoutingSlipVariableNames.Workflow.TemporalDirectory))
.AddStep(
ActivityNames.TransferFiles,
EndpointHelper.BuildExecuteActivityUri(TransferFilesToBucketActivity.ExecuteEndpointName),
TransferFilesToBucketActivity.ExecuteEndpointName,
new TransferFilesToBucketArguments(
message.DestinationBucket,
RoutingSlipVariableNames.Audio.TranscodedDirectory,
message.DestinationFolderKey));
routingSlipBuilder.AddActivity(
message.DestinationFolderKey))
.AddStep(
ActivityNames.UpdateTrackAudio,
EndpointHelper.BuildExecuteActivityUri(UpdateTrackAudioActivity.ExecuteEndpointName),
UpdateTrackAudioActivity.ExecuteEndpointName,
new UpdateTrackAudioArguments(
message.TrackId,
message.DestinationFolderKey,
RoutingSlipVariableNames.Audio.DurationSeconds));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackAudio);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder;
}
message.TrackId, message.DestinationFolderKey, RoutingSlipVariableNames.Audio.DurationSeconds))
.TrackFaults(message.TrackId, RoutingSlipVariableNames.ProcessKinds.TrackAudio);
}
@@ -1,11 +1,9 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.Tracks;
using Musify.Infrastructure.MassTransit.Activities.UploadIntents;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
@@ -13,57 +11,35 @@ public class CreateTrackRoutingSlipBuilder
{
public RoutingSlipBuilder Build(CreateTrackResourcesEvent message, Guid? correlationId)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
routingSlipBuilder.AddActivity(
var builder = RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.CopyPictureToFinal,
EndpointHelper.BuildExecuteActivityUri(CopyFileInBucketActivity.ExecuteEndpointName),
new CopyFileInBucketArguments(
message.Bucket,
message.PictureSourceKey,
message.Bucket,
message.PictureDestinationKey));
routingSlipBuilder.AddActivity(
CopyFileInBucketActivity.ExecuteEndpointName,
new CopyFileInBucketArguments(message.Bucket, message.PictureSourceKey, message.Bucket, message.PictureDestinationKey))
.AddStep(
ActivityNames.CopyAudioToFinal,
EndpointHelper.BuildExecuteActivityUri(CopyFileInBucketActivity.ExecuteEndpointName),
new CopyFileInBucketArguments(
message.Bucket,
message.AudioSourceKey,
message.Bucket,
message.AudioDestinationKey));
routingSlipBuilder.AddActivity(
CopyFileInBucketActivity.ExecuteEndpointName,
new CopyFileInBucketArguments(message.Bucket, message.AudioSourceKey, message.Bucket, message.AudioDestinationKey))
.AddStep(
ActivityNames.ConsumeUploadIntents,
EndpointHelper.BuildExecuteActivityUri(ConsumeUploadIntentsActivity.ExecuteEndpointName),
new ConsumeUploadIntentsArguments(
[message.PictureIntentId, message.AudioIntentId]));
routingSlipBuilder.AddActivity(
ConsumeUploadIntentsActivity.ExecuteEndpointName,
new ConsumeUploadIntentsArguments([message.PictureIntentId, message.AudioIntentId]))
.AddStep(
ActivityNames.PublishTrackProcessingEvents,
EndpointHelper.BuildExecuteActivityUri(PublishTrackProcessingEventsActivity.ExecuteEndpointName),
PublishTrackProcessingEventsActivity.ExecuteEndpointName,
new PublishTrackProcessingEventsArguments(
message.TrackId,
message.Bucket,
message.PictureDestinationKey,
message.AudioDestinationKey,
message.AudioProcessedFolderKey,
message.Sizes));
message.Sizes))
.TrackFaults(message.TrackId, RoutingSlipVariableNames.ProcessKinds.TrackCreation);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackCreation);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, message.Bucket);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, message.PictureDestinationKey);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.AudioKey, message.AudioDestinationKey);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
builder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, message.Bucket);
builder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, message.PictureDestinationKey);
builder.AddVariable(RoutingSlipVariableNames.Workflow.AudioKey, message.AudioDestinationKey);
return routingSlipBuilder;
return builder;
}
}
@@ -1,93 +0,0 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Microsoft.EntityFrameworkCore;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.Albums;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
public class DeleteAlbumRoutingSlipBuilder(
IDatabase database,
AlbumConfiguration albumConfiguration,
ApplicationStorageConfiguration storageConfiguration)
{
public async Task<RoutingSlipBuilder> BuildAsync(
Guid albumId,
long userId,
Guid? correlationId,
CancellationToken cancellationToken)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
var album = await database.Albums
.AsNoTracking()
.SingleOrDefaultAsync(p => p.Id == albumId, cancellationToken);
routingSlipBuilder.AddActivity(
ActivityNames.MarkAlbumAsRemoving,
EndpointHelper.BuildExecuteActivityUri(MarkAlbumAsRemovingActivity.ExecuteEndpointName),
new MarkAlbumAsRemovingArguments(albumId));
if (album?.Pictures != null)
{
var bucket = storageConfiguration.Bucket;
var routes = albumConfiguration.Routes;
if (album.Pictures.OriginalName is { } originalName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveAlbumOriginalPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildOriginalPicturePath(userId, originalName)));
}
if (album.Pictures.SmallName is { } smallName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveAlbumSmallPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildPicturePath(PictureSize.Small, smallName)));
}
if (album.Pictures.MediumName is { } mediumName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveAlbumMediumPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildPicturePath(PictureSize.Medium, mediumName)));
}
if (album.Pictures.LargeName is { } largeName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveAlbumLargePicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildPicturePath(PictureSize.Large, largeName)));
}
}
routingSlipBuilder.AddActivity(
ActivityNames.DeleteAlbumFromDb,
EndpointHelper.BuildExecuteActivityUri(DeleteAlbumFromDbActivity.ExecuteEndpointName),
new DeleteAlbumFromDbArguments(albumId));
return routingSlipBuilder;
}
}
@@ -1,93 +0,0 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Microsoft.EntityFrameworkCore;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.PlayLists;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
public class DeletePlayListRoutingSlipBuilder(
IDatabase database,
PlayListConfiguration playListConfiguration,
ApplicationStorageConfiguration storageConfiguration)
{
public async Task<RoutingSlipBuilder> BuildAsync(
Guid playListId,
long userId,
Guid? correlationId,
CancellationToken cancellationToken)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
var playList = await database.PlayLists
.AsNoTracking()
.SingleOrDefaultAsync(p => p.Id == playListId, cancellationToken);
routingSlipBuilder.AddActivity(
ActivityNames.MarkPlayListAsRemoving,
EndpointHelper.BuildExecuteActivityUri(MarkPlayListAsRemovingActivity.ExecuteEndpointName),
new MarkPlayListAsRemovingArguments(playListId));
if (playList?.Pictures != null)
{
var bucket = storageConfiguration.Bucket;
var routes = playListConfiguration.Routes;
if (playList.Pictures.OriginalName is { } originalName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemovePlayListOriginalPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildOriginalPicturePath(userId, originalName)));
}
if (playList.Pictures.SmallName is { } smallName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemovePlayListSmallPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildPicturePath(PictureSize.Small, smallName)));
}
if (playList.Pictures.MediumName is { } mediumName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemovePlayListMediumPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildPicturePath(PictureSize.Medium, mediumName)));
}
if (playList.Pictures.LargeName is { } largeName)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemovePlayListLargePicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
routes.BuildPicturePath(PictureSize.Large, largeName)));
}
}
routingSlipBuilder.AddActivity(
ActivityNames.DeletePlayListFromDb,
EndpointHelper.BuildExecuteActivityUri(DeletePlayListFromDbActivity.ExecuteEndpointName),
new DeletePlayListFromDbArguments(playListId));
return routingSlipBuilder;
}
}
@@ -0,0 +1,135 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Activities.LifeCycle;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
/// <summary>Marks an entity as removing, deletes its files from the bucket and finally deletes its row.</summary>
public class DeleteRoutingSlipBuilder(
IDatabase database,
TrackConfiguration trackConfiguration,
AlbumConfiguration albumConfiguration,
PlayListConfiguration playListConfiguration,
ApplicationStorageConfiguration storageConfiguration)
{
private sealed record PictureStepNames(string Original, string Small, string Medium, string Large);
private static readonly PictureStepNames AlbumSteps = new(
ActivityNames.RemoveAlbumOriginalPicture, ActivityNames.RemoveAlbumSmallPicture,
ActivityNames.RemoveAlbumMediumPicture, ActivityNames.RemoveAlbumLargePicture);
private static readonly PictureStepNames PlayListSteps = new(
ActivityNames.RemovePlayListOriginalPicture, ActivityNames.RemovePlayListSmallPicture,
ActivityNames.RemovePlayListMediumPicture, ActivityNames.RemovePlayListLargePicture);
public async Task<RoutingSlipBuilder> BuildTrackAsync(
Guid trackId, long userId, Guid? correlationId, CancellationToken cancellationToken)
{
var builder = RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.MarkTrackAsRemoving,
MarkTrackLifeCycleActivity.ExecuteEndpointName,
new MarkLifeCycleArguments(trackId, LifeCycleStatus.Removing));
var track = await database.Tracks.AsNoTracking().SingleOrDefaultAsync(t => t.Id == trackId, cancellationToken);
if (track != null)
{
var bucket = storageConfiguration.Bucket;
var routes = trackConfiguration.Routes;
if (!string.IsNullOrEmpty(track.Pictures.OriginalName))
{
builder.AddRemoveFile(
ActivityNames.RemoveTrackOriginalPicture, bucket,
routes.BuildOriginalPicturePath(userId, track.Pictures.OriginalName));
}
// Unprocessed tracks still point at the shared preset pictures, which must never be deleted.
if (track.Pictures.IsProcessed)
{
builder.AddRemoveFile(ActivityNames.RemoveTrackSmallPicture, bucket, routes.BuildPicturePath(PictureSize.Small, track.Pictures.SmallName));
builder.AddRemoveFile(ActivityNames.RemoveTrackMediumPicture, bucket, routes.BuildPicturePath(PictureSize.Medium, track.Pictures.MediumName));
builder.AddRemoveFile(ActivityNames.RemoveTrackLargePicture, bucket, routes.BuildPicturePath(PictureSize.Large, track.Pictures.LargeName));
}
if (!string.IsNullOrEmpty(track.Audio.OriginalName))
{
builder.AddRemoveFile(
ActivityNames.RemoveTrackOriginalAudio, bucket,
routes.BuildOriginalAudioPath(userId, track.Audio.OriginalName));
}
if (track.Audio.IsProcessed)
{
builder.AddRemoveFile(
ActivityNames.RemoveTrackProcessedAudio, bucket,
routes.BuildProcessedAudioPath(track.Audio.FolderName));
}
}
return builder.AddStep(
ActivityNames.DeleteTrackFromDb, DeleteTrackActivity.ExecuteEndpointName, new DeleteEntityArguments(trackId));
}
public async Task<RoutingSlipBuilder> BuildAlbumAsync(
Guid albumId, long userId, Guid? correlationId, CancellationToken cancellationToken)
{
var builder = RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.MarkAlbumAsRemoving,
MarkAlbumLifeCycleActivity.ExecuteEndpointName,
new MarkLifeCycleArguments(albumId, LifeCycleStatus.Removing));
var pictures = await database.Albums.AsNoTracking()
.Where(a => a.Id == albumId)
.Select(a => a.Pictures)
.SingleOrDefaultAsync(cancellationToken);
AddPictureRemovals(builder, AlbumSteps, albumConfiguration.Routes, userId, pictures);
return builder.AddStep(
ActivityNames.DeleteAlbumFromDb, DeleteAlbumActivity.ExecuteEndpointName, new DeleteEntityArguments(albumId));
}
public async Task<RoutingSlipBuilder> BuildPlayListAsync(
Guid playListId, long userId, Guid? correlationId, CancellationToken cancellationToken)
{
var builder = RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.MarkPlayListAsRemoving,
MarkPlayListLifeCycleActivity.ExecuteEndpointName,
new MarkLifeCycleArguments(playListId, LifeCycleStatus.Removing));
var pictures = await database.PlayLists.AsNoTracking()
.Where(p => p.Id == playListId)
.Select(p => p.Pictures)
.SingleOrDefaultAsync(cancellationToken);
AddPictureRemovals(builder, PlayListSteps, playListConfiguration.Routes, userId, pictures);
return builder.AddStep(
ActivityNames.DeletePlayListFromDb, DeletePlayListActivity.ExecuteEndpointName, new DeleteEntityArguments(playListId));
}
private void AddPictureRemovals(
RoutingSlipBuilder builder, PictureStepNames steps, PictureRoutes routes, long userId, EntityPictures? pictures)
{
if (pictures == null)
return;
var bucket = storageConfiguration.Bucket;
if (pictures.OriginalName is { } original)
builder.AddRemoveFile(steps.Original, bucket, routes.BuildOriginalPicturePath(userId, original));
if (pictures.SmallName is { } small)
builder.AddRemoveFile(steps.Small, bucket, routes.BuildPicturePath(PictureSize.Small, small));
if (pictures.MediumName is { } medium)
builder.AddRemoveFile(steps.Medium, bucket, routes.BuildPicturePath(PictureSize.Medium, medium));
if (pictures.LargeName is { } large)
builder.AddRemoveFile(steps.Large, bucket, routes.BuildPicturePath(PictureSize.Large, large));
}
}
@@ -1,106 +0,0 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Microsoft.EntityFrameworkCore;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.Tracks;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
public class DeleteTrackRoutingSlipBuilder(
IDatabase database,
TrackConfiguration trackConfiguration,
ApplicationStorageConfiguration storageConfiguration)
{
public async Task<RoutingSlipBuilder> BuildAsync(
Guid trackId,
long userId,
Guid? correlationId,
CancellationToken cancellationToken)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
var track = await database.Tracks
.AsNoTracking()
.SingleOrDefaultAsync(t => t.Id == trackId, cancellationToken);
routingSlipBuilder.AddActivity(
ActivityNames.MarkTrackAsRemoving,
EndpointHelper.BuildExecuteActivityUri(MarkTrackAsRemovingActivity.ExecuteEndpointName),
new MarkTrackAsRemovingArguments(trackId));
if (track != null)
{
var bucket = storageConfiguration.Bucket;
if (!string.IsNullOrEmpty(track.Pictures.OriginalName))
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveTrackOriginalPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
trackConfiguration.Routes.BuildOriginalPicturePath(userId, track.Pictures.OriginalName)));
}
if (track.Pictures.IsProcessed)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveTrackSmallPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
trackConfiguration.Routes.BuildPicturePath(PictureSize.Small, track.Pictures.SmallName)));
routingSlipBuilder.AddActivity(
ActivityNames.RemoveTrackMediumPicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
trackConfiguration.Routes.BuildPicturePath(PictureSize.Medium, track.Pictures.MediumName)));
routingSlipBuilder.AddActivity(
ActivityNames.RemoveTrackLargePicture,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
trackConfiguration.Routes.BuildPicturePath(PictureSize.Large, track.Pictures.LargeName)));
}
if (!string.IsNullOrEmpty(track.Audio.OriginalName))
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveTrackOriginalAudio,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
trackConfiguration.Routes.BuildOriginalAudioPath(userId, track.Audio.OriginalName)));
}
if (track.Audio.IsProcessed)
{
routingSlipBuilder.AddActivity(
ActivityNames.RemoveTrackProcessedAudio,
EndpointHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
bucket,
trackConfiguration.Routes.BuildProcessedAudioPath(track.Audio.FolderName)));
}
}
routingSlipBuilder.AddActivity(
ActivityNames.DeleteTrackFromDb,
EndpointHelper.BuildExecuteActivityUri(DeleteTrackFromDbActivity.ExecuteEndpointName),
new DeleteTrackFromDbArguments(trackId));
return routingSlipBuilder;
}
}
@@ -0,0 +1,74 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.MassTransit.Activities.Albums;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.PlayLists;
using Musify.Infrastructure.MassTransit.Activities.UploadIntents;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
/// <summary>
/// Moves an uploaded album/playlist picture to its final key, consumes the upload intent and hands over to the picture workflow.
/// Creation faults mark the whole entity as failed; update faults only report the picture as failed.
/// </summary>
public class PictureSourceRoutingSlipBuilder
{
public RoutingSlipBuilder Build(CreateAlbumResourcesEvent message, Guid? correlationId) =>
BuildSlip(
message.AlbumId, message.PictureIntentId, message.Bucket, message.PictureSourceKey, message.PictureDestinationKey,
correlationId, RoutingSlipVariableNames.ProcessKinds.AlbumCreation,
ActivityNames.PublishAlbumPictureProcessingEvent, PublishAlbumPictureProcessingEventActivity.ExecuteEndpointName,
new PublishAlbumPictureProcessingEventArguments(message.AlbumId, message.Bucket, message.PictureDestinationKey, message.Sizes));
public RoutingSlipBuilder Build(UpdateAlbumPictureSourceEvent message, Guid? correlationId) =>
BuildSlip(
message.AlbumId, message.PictureIntentId, message.Bucket, message.PictureSourceKey, message.PictureDestinationKey,
correlationId, RoutingSlipVariableNames.ProcessKinds.AlbumPicture,
ActivityNames.PublishAlbumPictureProcessingEvent, PublishAlbumPictureProcessingEventActivity.ExecuteEndpointName,
new PublishAlbumPictureProcessingEventArguments(message.AlbumId, message.Bucket, message.PictureDestinationKey, message.Sizes));
public RoutingSlipBuilder Build(CreatePlayListResourcesEvent message, Guid? correlationId) =>
BuildSlip(
message.PlayListId, message.PictureIntentId, message.Bucket, message.PictureSourceKey, message.PictureDestinationKey,
correlationId, RoutingSlipVariableNames.ProcessKinds.PlayListCreation,
ActivityNames.PublishPlayListPictureProcessingEvent, PublishPlayListPictureProcessingEventActivity.ExecuteEndpointName,
new PublishPlayListPictureProcessingEventArguments(message.PlayListId, message.Bucket, message.PictureDestinationKey, message.Sizes));
public RoutingSlipBuilder Build(UpdatePlayListPictureSourceEvent message, Guid? correlationId) =>
BuildSlip(
message.PlayListId, message.PictureIntentId, message.Bucket, message.PictureSourceKey, message.PictureDestinationKey,
correlationId, RoutingSlipVariableNames.ProcessKinds.PlayListPicture,
ActivityNames.PublishPlayListPictureProcessingEvent, PublishPlayListPictureProcessingEventActivity.ExecuteEndpointName,
new PublishPlayListPictureProcessingEventArguments(message.PlayListId, message.Bucket, message.PictureDestinationKey, message.Sizes));
private static RoutingSlipBuilder BuildSlip(
Guid subjectId,
Guid pictureIntentId,
string bucket,
string pictureSourceKey,
string pictureDestinationKey,
Guid? correlationId,
string processKind,
string publishStepName,
string publishEndpointName,
object publishArguments)
{
var builder = RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.CopyPictureToFinal,
CopyFileInBucketActivity.ExecuteEndpointName,
new CopyFileInBucketArguments(bucket, pictureSourceKey, bucket, pictureDestinationKey))
.AddStep(
ActivityNames.ConsumeUploadIntents,
ConsumeUploadIntentsActivity.ExecuteEndpointName,
new ConsumeUploadIntentsArguments([pictureIntentId]))
.AddStep(publishStepName, publishEndpointName, publishArguments)
.TrackFaults(subjectId, processKind);
builder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket);
builder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey);
return builder;
}
}
@@ -1,176 +1,86 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Application.Shared;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.Pictures;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
/// <summary>Downloads a picture, resizes it to the three sizes, uploads them and records the result on the owner.</summary>
public class PictureWorkflowRoutingSlipBuilder(WorkerConfiguration workerConfiguration)
{
public RoutingSlipBuilder Build(UpdateTrackPictureEvent message, Guid? correlationId)
{
var routingSlipBuilder = BuildPictureWorkflow(
message.Bucket,
message.SourceKey,
message.Sizes,
correlationId);
public RoutingSlipBuilder Build(UpdateTrackPictureEvent message, Guid? correlationId) =>
BuildPictureWorkflow(
message.TrackId, message.Bucket, message.SourceKey, message.Sizes, correlationId,
ActivityNames.UpdateTrackPicture, UpdateTrackPictureActivity.ExecuteEndpointName,
RoutingSlipVariableNames.ProcessKinds.TrackPicture);
routingSlipBuilder.AddActivity(
ActivityNames.UpdateTrackPicture,
EndpointHelper.BuildExecuteActivityUri(UpdateTrackPictureActivity.ExecuteEndpointName),
new UpdateTrackPictureArguments(
message.TrackId,
message.SourceKey,
RoutingSlipVariableNames.Picture.SmallResizedFilePath,
RoutingSlipVariableNames.Picture.MediumResizedFilePath,
RoutingSlipVariableNames.Picture.LargeResizedFilePath));
public RoutingSlipBuilder Build(UpdatePlayListPictureEvent message, Guid? correlationId) =>
BuildPictureWorkflow(
message.PlayListId, message.Bucket, message.SourceKey, message.Sizes, correlationId,
ActivityNames.UpdatePlayListPicture, UpdatePlayListPictureActivity.ExecuteEndpointName,
RoutingSlipVariableNames.ProcessKinds.PlayListPicture);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.TrackId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.TrackPicture);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder;
}
public RoutingSlipBuilder Build(UpdatePlayListPictureEvent message, Guid? correlationId)
{
var routingSlipBuilder = BuildPictureWorkflow(
message.Bucket,
message.SourceKey,
message.Sizes,
correlationId);
routingSlipBuilder.AddActivity(
ActivityNames.UpdatePlayListPicture,
EndpointHelper.BuildExecuteActivityUri(UpdatePlayListPictureActivity.ExecuteEndpointName),
new UpdatePlayListPictureArguments(
message.PlayListId,
message.SourceKey,
RoutingSlipVariableNames.Picture.SmallResizedFilePath,
RoutingSlipVariableNames.Picture.MediumResizedFilePath,
RoutingSlipVariableNames.Picture.LargeResizedFilePath));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.PlayListId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.PlayListPicture);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder;
}
public RoutingSlipBuilder Build(UpdateAlbumPictureEvent message, Guid? correlationId)
{
var routingSlipBuilder = BuildPictureWorkflow(
message.Bucket,
message.SourceKey,
message.Sizes,
correlationId);
routingSlipBuilder.AddActivity(
ActivityNames.UpdateAlbumPicture,
EndpointHelper.BuildExecuteActivityUri(UpdateAlbumPictureActivity.ExecuteEndpointName),
new UpdateAlbumPictureArguments(
message.AlbumId,
message.SourceKey,
RoutingSlipVariableNames.Picture.SmallResizedFilePath,
RoutingSlipVariableNames.Picture.MediumResizedFilePath,
RoutingSlipVariableNames.Picture.LargeResizedFilePath));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, message.AlbumId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, RoutingSlipVariableNames.ProcessKinds.AlbumPicture);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder;
}
public RoutingSlipBuilder Build(UpdateAlbumPictureEvent message, Guid? correlationId) =>
BuildPictureWorkflow(
message.AlbumId, message.Bucket, message.SourceKey, message.Sizes, correlationId,
ActivityNames.UpdateAlbumPicture, UpdateAlbumPictureActivity.ExecuteEndpointName,
RoutingSlipVariableNames.ProcessKinds.AlbumPicture);
private RoutingSlipBuilder BuildPictureWorkflow(
Guid subjectId,
string bucket,
string sourceKey,
ImageSizes sizes,
Guid? correlationId)
Guid? correlationId,
string updateStepName,
string updateEndpointName,
string processKind)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
routingSlipBuilder.AddActivity(
var builder = RoutingSlips.Create(correlationId)
.AddStep(
ActivityNames.GeneratePictureWorkflowPaths,
EndpointHelper.BuildExecuteActivityUri(GeneratePictureWorkflowPathsActivity.ExecuteEndpointName),
new GeneratePictureWorkflowPathsArguments(
workerConfiguration.Routes.TemporaryFilesDirectory,
sourceKey));
routingSlipBuilder.AddActivity(
GeneratePictureWorkflowPathsActivity.ExecuteEndpointName,
new GeneratePictureWorkflowPathsArguments(workerConfiguration.Routes.TemporaryFilesDirectory, sourceKey))
.AddStep(
ActivityNames.DownloadFile,
EndpointHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
bucket,
DownloadFileFromBucketActivity.ExecuteEndpointName,
new DownloadFileFromBucketArguments(bucket, sourceKey, RoutingSlipVariableNames.Picture.OriginalFilePath));
AddResize(builder, ActivityNames.ResizeSmall, ActivityNames.UploadSmall, bucket, RoutingSlipVariableNames.Picture.SmallResizedFilePath, sizes.Small);
AddResize(builder, ActivityNames.ResizeMedium, ActivityNames.UploadMedium, bucket, RoutingSlipVariableNames.Picture.MediumResizedFilePath, sizes.Medium);
AddResize(builder, ActivityNames.ResizeLarge, ActivityNames.UploadLarge, bucket, RoutingSlipVariableNames.Picture.LargeResizedFilePath, sizes.Large);
return builder
.AddStep(
updateStepName,
updateEndpointName,
new UpdatePicturesArguments(
subjectId,
sourceKey,
RoutingSlipVariableNames.Picture.OriginalFilePath));
AddResizeActivity(
routingSlipBuilder,
ActivityNames.ResizeSmall,
ActivityNames.UploadSmall,
bucket,
RoutingSlipVariableNames.Picture.SmallResizedFilePath,
sizes.Small);
AddResizeActivity(
routingSlipBuilder,
ActivityNames.ResizeMedium,
ActivityNames.UploadMedium,
bucket,
RoutingSlipVariableNames.Picture.MediumResizedFilePath,
sizes.Medium);
AddResizeActivity(
routingSlipBuilder,
ActivityNames.ResizeLarge,
ActivityNames.UploadLarge,
bucket,
RoutingSlipVariableNames.Picture.LargeResizedFilePath,
sizes.Large);
return routingSlipBuilder;
RoutingSlipVariableNames.Picture.LargeResizedFilePath))
.TrackFaults(subjectId, processKind);
}
private static void AddResizeActivity(
RoutingSlipBuilder routingSlipBuilder,
string resizeActivityName,
string uploadActivityName,
string bucketName,
string destinationFilePathVariableName,
ImageSize imageSize)
{
routingSlipBuilder.AddActivity(
resizeActivityName,
EndpointHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName),
private static void AddResize(
RoutingSlipBuilder builder,
string resizeStepName,
string uploadStepName,
string bucket,
string resizedFileVariable,
ImageSize size) =>
builder
.AddStep(
resizeStepName,
ResizePictureActivity.ExecuteEndpointName,
new ResizePictureLocalArguments(
RoutingSlipVariableNames.Picture.OriginalFilePath,
destinationFilePathVariableName,
imageSize.Width,
imageSize.Height));
routingSlipBuilder.AddActivity(
uploadActivityName,
EndpointHelper.BuildExecuteActivityUri(UploadFileToBucketActivity.ExecuteEndpointName),
new UploadFileToBucketArguments(
destinationFilePathVariableName,
bucketName,
imageSize.SavePath));
}
RoutingSlipVariableNames.Picture.OriginalFilePath, resizedFileVariable, size.Width, size.Height))
.AddStep(
uploadStepName,
UploadFileToBucketActivity.ExecuteEndpointName,
new UploadFileToBucketArguments(resizedFileVariable, bucket, size.SavePath));
}
@@ -1,87 +0,0 @@
using MassTransit;
using MassTransit.Courier.Contracts;
using Musify.Application.Events;
using Musify.Application.Shared;
using Musify.Infrastructure.MassTransit.Activities.Files;
using Musify.Infrastructure.MassTransit.Activities.PlayLists;
using Musify.Infrastructure.MassTransit.Activities.UploadIntents;
using Musify.Infrastructure.MassTransit.Arguments;
using Musify.Infrastructure.MassTransit.Consumers;
namespace Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
public class PlayListPictureSourceRoutingSlipBuilder
{
public RoutingSlipBuilder Build(CreatePlayListResourcesEvent message, Guid? correlationId)
=> BuildSlip(
message.PlayListId,
message.PictureIntentId,
message.Bucket,
message.PictureSourceKey,
message.PictureDestinationKey,
message.Sizes,
correlationId,
RoutingSlipVariableNames.ProcessKinds.PlayListCreation);
public RoutingSlipBuilder Build(UpdatePlayListPictureSourceEvent message, Guid? correlationId)
=> BuildSlip(
message.PlayListId,
message.PictureIntentId,
message.Bucket,
message.PictureSourceKey,
message.PictureDestinationKey,
message.Sizes,
correlationId,
RoutingSlipVariableNames.ProcessKinds.PlayListPicture);
private static RoutingSlipBuilder BuildSlip(
Guid playListId,
Guid pictureIntentId,
string bucket,
string pictureSourceKey,
string pictureDestinationKey,
ImageSizes sizes,
Guid? correlationId,
string processKind)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.CorrelationId, correlationId ?? Guid.Empty);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(RoutingSlipCleanUpConsumer.QueueName),
RoutingSlipEvents.Completed | RoutingSlipEvents.Faulted);
routingSlipBuilder.AddActivity(
ActivityNames.CopyPictureToFinal,
EndpointHelper.BuildExecuteActivityUri(CopyFileInBucketActivity.ExecuteEndpointName),
new CopyFileInBucketArguments(
bucket,
pictureSourceKey,
bucket,
pictureDestinationKey));
routingSlipBuilder.AddActivity(
ActivityNames.ConsumeUploadIntents,
EndpointHelper.BuildExecuteActivityUri(ConsumeUploadIntentsActivity.ExecuteEndpointName),
new ConsumeUploadIntentsArguments([pictureIntentId]));
routingSlipBuilder.AddActivity(
ActivityNames.PublishPlayListPictureProcessingEvent,
EndpointHelper.BuildExecuteActivityUri(PublishPlayListPictureProcessingEventActivity.ExecuteEndpointName),
new PublishPlayListPictureProcessingEventArguments(
playListId,
bucket,
pictureDestinationKey,
sizes));
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.SubjectId, playListId);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.ProcessKind, processKind);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.Bucket, bucket);
routingSlipBuilder.AddVariable(RoutingSlipVariableNames.Workflow.PictureKey, pictureDestinationKey);
routingSlipBuilder.AddSubscription(
EndpointHelper.BuildConsumerUri(ProcessingSlipFaultConsumer.QueueName),
RoutingSlipEvents.Faulted);
return routingSlipBuilder;
}
}
@@ -1,4 +1,4 @@
namespace Musify.Infrastructure.MassTransit;
namespace Musify.Infrastructure.MassTransit.RoutingSlip;
internal static class RoutingSlipVariableNames
{
@@ -0,0 +1,48 @@
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;
/// <summary>Shared building blocks for every routing slip in the worker.</summary>
internal static class RoutingSlips
{
public static Uri ExecuteUri(string endpointName) => new($"queue:{endpointName}_execute");
public static Uri ConsumerUri(string queueName) => new($"queue:{queueName}");
/// <summary>A new slip that carries the correlation id and always cleans its temporary directory up.</summary>
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;
}
/// <summary>Publishes a failure event for the subject when the slip faults.</summary>
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));
}