NotS1mp/Musify
0
0
Fork 0
Code Issues 23 Pull Requests Actions Packages Projects Releases Wiki Activity

Changes

Browse Source
This commit is contained in:
Admin
2026-03-24 14:45:20 +01:00
parent da4b76f499
commit 27d68986c5
66 changed files with 1830 additions and 922 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Infrastructure/Messaging/Activities/Arguments/CopyFileArguments.cs
-8
View File
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record CopyFileArguments(
string SourceBucketName,
string SourceKeyName,
string DestinationBucketName,
string DestinationKeyName);
}
Infrastructure/Messaging/Activities/Arguments/DownloadFileFromBucketArguments.cs
-7
View File
@@ -1,7 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record DownloadFileFromBucketArguments(
string BucketName,
string KeyName,
string DestinationPath);
}
Infrastructure/Messaging/Activities/Arguments/RemoveFileFromBucketArguments.cs
-6
View File
@@ -1,6 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record RemoveFileFromBucketArguments(
string BucketName,
string KeyName);
}
Infrastructure/Messaging/Activities/Arguments/ResizePictureArgument.cs
-10
View File
@@ -1,10 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record ResizePictureArgument(
string OriginalBucketName,
string OriginalKeyName,
string DestinationBucketName,
string DestinationKeyName,
int Width,
int Height);
}
Infrastructure/Messaging/Activities/Arguments/ResizePictureLocalArguments.cs
-8
View File
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record ResizePictureLocalArguments(
string SourceFilePath,
string DestinationFilePath,
int Width,
int Height);
}
Infrastructure/Messaging/Activities/Arguments/TranscodeDashAudioArguments.cs
-4
View File
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record TranscodeDashAudioArguments();
}
Infrastructure/Messaging/Activities/Arguments/TransferFilesToBucketArguments.cs
-6
View File
@@ -1,6 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record TransferFilesToBucketArguments(
string DestinationBucketName,
string DestinationKeyName);
}
Infrastructure/Messaging/Activities/Arguments/UpdatePlayListPictureArguments.cs
-8
View File
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record UpdatePlayListPictureArguments(
Guid PlayListId,
string SmallPictureKeyName,
string MediumPictureKeyName,
string LargePictureKeyName);
}
Infrastructure/Messaging/Activities/Arguments/UpdateTrackPictureArguments.cs
-8
View File
@@ -1,8 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record UpdateTrackPictureArguments(
Guid TrackId,
string SmallPictureKeyName,
string MediumPictureKeyName,
string LargePictureKeyName);
}
Infrastructure/Messaging/Activities/DownloadFileFromBucketActivity.cs
-80
View File
@@ -1,80 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class DownloadFileFromBucketActivity(
IStorageHandler storageHandler,
ILogger<DownloadFileFromBucketActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<DownloadFileFromBucketArguments>
{
public const string ExecuteEndpointName = "Download-File-From-Bucket";
public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(DownloadFileFromBucketActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
var getFile = await storageHandler.GetFileAsync(
executeContext.Arguments.BucketName,
executeContext.Arguments.KeyName,
executeContext.CancellationToken);
if (!getFile.IsSuccess)
{
var errorMessage = string.Join("; ", getFile.Errors);
logger.LogWarning("Failed to download file from bucket. Bucket: {BucketName}, Key: {KeyName}, Errors: {Errors}",
executeContext.Arguments.BucketName,
executeContext.Arguments.KeyName,
errorMessage);
throw new Exception(errorMessage);
}
using var fileStream = getFile.Value;
if (fileStream.CanSeek)
fileStream.Position = 0;
var fileName = Path.GetFileName(executeContext.Arguments.KeyName);
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine(executeContext.Arguments.DestinationPath, folderName);
var destinationPath = Path.Combine(workingDirectory, fileName);
Directory.CreateDirectory(workingDirectory);
using var destinationStream = File.Create(destinationPath);
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.CompletedWithVariables(new
{
SourceFilePath = destinationPath,
WorkingDirectory = workingDirectory,
DestinationFolderName = folderName
} );
}
catch (Exception exception)
{
logger.LogError(exception, "Error downloading file from bucket");
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Activities/GenerateAudioWorkflowPathsActivity.cs
View File
Infrastructure/Messaging/Activities/GeneratePictureWorkflowPathsActivity.cs
View File
Infrastructure/Messaging/Activities/Logs/DownloadFileFromBucketLog.cs
-7
View File
@@ -1,7 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Logs
{
public record DownloadFileFromBucketLog(
string BucketName,
string KeyName,
string DestinationPath);
}
Infrastructure/Messaging/Activities/Logs/TranscodeDashAudioLog.cs
-4
View File
@@ -1,4 +0,0 @@
namespace Musify.Infrastructure.Messaging.Activities.Logs
{
public record TranscodeDashAudioLog(string WorkingDirectory);
}
Infrastructure/Messaging/Activities/RemoveFileFromBucketActivity.cs
-62
View File
@@ -1,62 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class RemoveFileFromBucketActivity(
IStorageHandler storageHandler,
ILogger<RemoveFileFromBucketActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<RemoveFileFromBucketArguments>
{
public const string ExecuteEndpointName = "Remove-File";
public async Task<ExecutionResult> Execute(ExecuteContext<RemoveFileFromBucketArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(RemoveFileFromBucketActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
var removeFile = await storageHandler.RemoveFileAsync(
executeContext.Arguments.BucketName,
executeContext.Arguments.KeyName,
executeContext.CancellationToken);
if (!removeFile.IsSuccess)
{
var errorMessage = string.Join("; ", removeFile.Errors);
logger.LogWarning("Failed to remove file {KeyName} from bucket {BucketName}. Errors: {Errors}",
executeContext.Arguments.KeyName,
executeContext.Arguments.BucketName,
errorMessage);
}
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error removing file {KeyName} from bucket {BucketName}",
executeContext.Arguments.KeyName,
executeContext.Arguments.BucketName);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Activities/ResizePictureActivity.cs
-87
View File
@@ -1,87 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class ResizePictureActivity(
IPictureHandler pictureHandler,
ILogger<ResizePictureActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<ResizePictureLocalArguments>
{
public const string ExecuteEndpointName = "Resize-Picture";
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(ResizePictureActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
if (!File.Exists(executeContext.Arguments.SourceFilePath))
{
logger.LogWarning("Source file not found at {SourceFilePath}", executeContext.Arguments.SourceFilePath);
throw new FileNotFoundException($"Source file not found: {executeContext.Arguments.SourceFilePath}");
}
await using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath);
var resizedPicture = await pictureHandler.ResizePictureAsync(
fileStream,
executeContext.Arguments.Width,
executeContext.Arguments.Height,
executeContext.CancellationToken);
if (!resizedPicture.IsSuccess)
{
var errorMessage = string.Join("; ", resizedPicture.Errors);
logger.LogError("Failed to resize picture {SourceFilePath} to {Width}x{Height}. Errors: {Errors}",
executeContext.Arguments.SourceFilePath,
executeContext.Arguments.Width,
executeContext.Arguments.Height,
errorMessage);
throw new Exception(errorMessage);
}
var destinationDirectory = Path.GetDirectoryName(executeContext.Arguments.DestinationFilePath);
if (!string.IsNullOrEmpty(destinationDirectory))
{
Directory.CreateDirectory(destinationDirectory);
}
await using var destinationStream = File.Create(executeContext.Arguments.DestinationFilePath);
await resizedPicture.Value.CopyToAsync(destinationStream, executeContext.CancellationToken);
logger.LogDebug("Picture resized from {SourceFilePath} to {DestinationFilePath} ({Width}x{Height})",
executeContext.Arguments.SourceFilePath,
executeContext.Arguments.DestinationFilePath,
executeContext.Arguments.Width,
executeContext.Arguments.Height);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error resizing picture {SourceFilePath}",
executeContext.Arguments.SourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Activities/TranscodeDashAudioActivity.cs
-70
View File
@@ -1,70 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class TranscodeDashAudioActivity(
IAudioTranscoder audioTranscoder,
ILogger<TranscodeDashAudioActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<TranscodeDashAudioArguments>
{
public const string ExecuteEndpointName = "Transcode-Dash-Audio";
public async Task<ExecutionResult> Execute(ExecuteContext<TranscodeDashAudioArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(TranscodeDashAudioActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var sourceFilePath = executeContext.GetVariable<string>("SourceFilePath");
var workingDirectory = executeContext.GetVariable<string>("WorkingDirectory");
if (string.IsNullOrWhiteSpace(sourceFilePath) || string.IsNullOrWhiteSpace(workingDirectory))
{
throw new InvalidOperationException("Transcode activity requires SourceFilePath and WorkingDirectory variables.");
}
try
{
await using var fileStream = File.OpenRead(sourceFilePath);
var result = await audioTranscoder.TranscodeToDashAsync(
fileStream,
workingDirectory,
executeContext.CancellationToken);
if (!result.IsSuccess)
{
var errorMessage = string.Join("; ", result.Errors);
logger.LogError("Transcoding failed for file {SourceFilePath}. Errors: {ErrorMessage}",
sourceFilePath, errorMessage);
throw new Exception(errorMessage);
}
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error during transcoding activity for file {SourceFilePath}",
sourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Activities/TransferFilesToBucket.cs
-71
View File
@@ -1,71 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class TransferFilesToBucket(
IStorageHandler storageHandler,
ILogger<TransferFilesToBucket> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<TransferFilesToBucketArguments>
{
public const string ExecuteEndpointName = "Transfer-Files-To-Bucket";
public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(TransferFilesToBucket),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
var folderPath = executeContext.GetVariable<string>("WorkingDirectory");
if (string.IsNullOrWhiteSpace(folderPath))
{
throw new InvalidOperationException("Transfer activity requires WorkingDirectory variable.");
}
try
{
var transferFilesResult = await storageHandler.TransferFilesAsync(
folderPath,
executeContext.Arguments.DestinationBucketName,
executeContext.Arguments.DestinationKeyName,
executeContext.CancellationToken);
if (!transferFilesResult.IsSuccess)
{
var errorMessage = string.Join("; ", transferFilesResult.Errors);
logger.LogWarning("Failed to transfer files from {FolderPath} to bucket {DestinationBucketName}, key {DestinationKeyName}. Errors: {Errors}",
folderPath,
executeContext.Arguments.DestinationBucketName,
executeContext.Arguments.DestinationKeyName,
errorMessage);
throw new Exception(errorMessage);
}
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error transferring files from {FolderPath} to bucket {DestinationBucketName}",
folderPath,
executeContext.Arguments.DestinationBucketName);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Activities/UpdatePlayListPictureActivity.cs
-68
View File
@@ -1,68 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class UpdatePlayListPictureActivity(
IDatabase database,
ILogger<UpdatePlayListPictureActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<UpdatePlayListPictureArguments>
{
public const string ExecuteEndpointName = "Update-PlayList-Picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(UpdatePlayListPictureActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
var playList = await database.PlayLists.FindAsync(
[executeContext.Arguments.PlayListId],
cancellationToken: executeContext.CancellationToken);
if (playList is null)
{
logger.LogWarning("PlayList with id {PlayListId} not found",
executeContext.Arguments.PlayListId);
throw new InvalidOperationException($"PlayList with id {executeContext.Arguments.PlayListId} not found");
}
playList.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName;
playList.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName;
playList.LargePictureKeyName = executeContext.Arguments.LargePictureKeyName;
database.PlayLists.Update(playList);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated PlayList {PlayListId} pictures",
executeContext.Arguments.PlayListId);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error updating PlayList {PlayListId} pictures",
executeContext.Arguments.PlayListId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Activities/UpdateTrackPictureActivity.cs
-68
View File
@@ -1,68 +0,0 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class UpdateTrackPictureActivity(
IDatabase database,
ILogger<UpdateTrackPictureActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<UpdateTrackPictureArguments>
{
public const string ExecuteEndpointName = "Update-Track-Picture";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateTrackPictureArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(UpdateTrackPictureActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
var track = await database.Tracks.FindAsync(
[executeContext.Arguments.TrackId],
cancellationToken: executeContext.CancellationToken);
if (track is null)
{
logger.LogWarning("Track with id {TrackId} not found",
executeContext.Arguments.TrackId);
throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found");
}
track.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName;
track.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName;
track.LargePictureKeyName = executeContext.Arguments.LargePictureKeyName;
database.Tracks.Update(track);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated Track {TrackId} pictures",
executeContext.Arguments.TrackId);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error updating Track {TrackId} pictures",
executeContext.Arguments.TrackId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
Infrastructure/Messaging/Consumers/MessagingHelper.cs
-7
View File
@@ -1,7 +0,0 @@
namespace Musify.Infrastructure.Messaging.Consumers
{
internal static class MessagingHelper
{
internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{endpointName}_execute");
}
}
Infrastructure/Messaging/Consumers/RemoveFileConsumer.cs
-28
View File
@@ -1,28 +0,0 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class RemoveFileConsumer(IBus bus)
: IConsumer<RemoveFileEvent>
{
public const string QueueName = "Remove-File";
public async Task Consume(ConsumeContext<RemoveFileEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddActivity(
"RemoveFile",
MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
consumeContext.Message.BucketName,
consumeContext.Message.KeyName));
var routingSlip = routingSlipBuilder.Build();
await bus.Execute(routingSlip);
}
}
}
Infrastructure/Messaging/Consumers/TranscodeAudioFromTrackConsumer.cs
-44
View File
@@ -1,44 +0,0 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class TranscodeAudioFromTrackConsumer(
IBus bus,
WorkerConfiguration workerConfiguration)
: IConsumer<TranscodeAudioFromTrackEvent>
{
public const string QueueName = "Transcode-Audio-From-Track";
public async Task Consume(ConsumeContext<TranscodeAudioFromTrackEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
routingSlipBuilder.AddActivity(
"DownloadFile",
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName,
workerConfiguration.Routes.TemporaryFilesDirectory));
routingSlipBuilder.AddActivity(
"TranscodeAudio",
MessagingHelper.BuildExecuteActivityUri(TranscodeDashAudioActivity.ExecuteEndpointName),
new TranscodeDashAudioArguments());
routingSlipBuilder.AddActivity(
"UploadFile",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments(
consumeContext.Message.DestinationBucketName,
consumeContext.Message.DestinationKeyName));
var routingSlip = routingSlipBuilder.Build();
await bus.Execute(routingSlip);
}
}
}
Infrastructure/Messaging/Consumers/UpdatePlayListPictureConsumer.cs
-112
View File
@@ -1,112 +0,0 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class UpdatePlayListPictureConsumer(
IBus bus,
WorkerConfiguration workerConfiguration)
: IConsumer<UpdatePlayListPictureEvent>
{
public const string QueueName = "Update-PlayList-Picture";
public async Task Consume(ConsumeContext<UpdatePlayListPictureEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine(workerConfiguration.Routes.TemporaryFilesDirectory, folderName);
routingSlipBuilder.AddActivity(
"DownloadFile",
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName,
workingDirectory));
AddResizeActivity(
routingSlipBuilder,
"ResizeSmall",
consumeContext.Message.SmallPictureKeyName,
consumeContext.Message.SmallPictureWidth,
consumeContext.Message.SmallPictureHeight,
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
"ResizeMedium",
consumeContext.Message.MediumPictureKeyName,
consumeContext.Message.MediumPictureWidth,
consumeContext.Message.MediumPictureHeight,
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
"ResizeLarge",
consumeContext.Message.LargePictureKeyName,
consumeContext.Message.LargePictureWidth,
consumeContext.Message.LargePictureHeight,
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
routingSlipBuilder.AddActivity(
"DeleteOriginal",
MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName));
routingSlipBuilder.AddActivity(
"UploadFiles",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments(
consumeContext.Message.SourceBucketName,
folderName));
routingSlipBuilder.AddActivity(
"UpdatePlayListPicture",
MessagingHelper.BuildExecuteActivityUri(UpdatePlayListPictureActivity.ExecuteEndpointName),
new UpdatePlayListPictureArguments(
consumeContext.Message.PlayListId,
consumeContext.Message.SmallPictureKeyName,
consumeContext.Message.MediumPictureKeyName,
consumeContext.Message.LargePictureKeyName));
var routingSlip = routingSlipBuilder.Build();
await bus.Execute(routingSlip);
}
void AddResizeActivity(
RoutingSlipBuilder routingSlipBuilder,
string activityName,
string destinationKeyName,
int width,
int height,
string originalPictureKeyName,
string bucketName,
string workingDirectory)
{
var fileName = Path.GetFileName(originalPictureKeyName);
var sourceFilePath = Path.Combine(workingDirectory, fileName);
var destinationFileName = Path.GetFileNameWithoutExtension(destinationKeyName) + Path.GetExtension(fileName);
var destinationFilePath = Path.Combine(workingDirectory, destinationFileName);
routingSlipBuilder.AddActivity(
activityName,
MessagingHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName),
new ResizePictureLocalArguments(
sourceFilePath,
destinationFilePath,
width,
height));
}
}
}
Infrastructure/Messaging/Consumers/UpdateTrackPictureConsumer.cs
-112
View File
@@ -1,112 +0,0 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class UpdateTrackPictureConsumer(
IBus bus,
WorkerConfiguration workerConfiguration)
: IConsumer<UpdateTrackPictureEvent>
{
public const string QueueName = "Update-Track-Picture";
public async Task Consume(ConsumeContext<UpdateTrackPictureEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine(workerConfiguration.Routes.TemporaryFilesDirectory, folderName);
routingSlipBuilder.AddActivity(
"DownloadFile",
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName,
workingDirectory));
AddResizeActivity(
routingSlipBuilder,
"ResizeSmall",
consumeContext.Message.SmallPictureKeyName,
consumeContext.Message.SmallPictureWidth,
consumeContext.Message.SmallPictureHeight,
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
"ResizeMedium",
consumeContext.Message.MediumPictureKeyName,
consumeContext.Message.MediumPictureWidth,
consumeContext.Message.MediumPictureHeight,
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
"ResizeLarge",
consumeContext.Message.LargePictureKeyName,
consumeContext.Message.LargePictureWidth,
consumeContext.Message.LargePictureHeight,
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
routingSlipBuilder.AddActivity(
"DeleteOriginal",
MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName));
routingSlipBuilder.AddActivity(
"UploadFiles",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments(
consumeContext.Message.SourceBucketName,
folderName));
routingSlipBuilder.AddActivity(
"UpdateTrackPicture",
MessagingHelper.BuildExecuteActivityUri(UpdateTrackPictureActivity.ExecuteEndpointName),
new UpdateTrackPictureArguments(
consumeContext.Message.TrackId,
consumeContext.Message.SmallPictureKeyName,
consumeContext.Message.MediumPictureKeyName,
consumeContext.Message.LargePictureKeyName));
var routingSlip = routingSlipBuilder.Build();
await bus.Execute(routingSlip);
}
void AddResizeActivity(
RoutingSlipBuilder routingSlipBuilder,
string activityName,
string destinationKeyName,
int width,
int height,
string originalPictureKeyName,
string bucketName,
string workingDirectory)
{
var fileName = Path.GetFileName(originalPictureKeyName);
var sourceFilePath = Path.Combine(workingDirectory, fileName);
var destinationFileName = Path.GetFileNameWithoutExtension(destinationKeyName) + Path.GetExtension(fileName);
var destinationFilePath = Path.Combine(workingDirectory, destinationFileName);
routingSlipBuilder.AddActivity(
activityName,
MessagingHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName),
new ResizePictureLocalArguments(
sourceFilePath,
destinationFilePath,
width,
height));
}
}
}
Infrastructure/Messaging/Filters/ProcessTrackingConsumeFilter.cs
-48
View File
@@ -1,48 +0,0 @@
using MassTransit;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
namespace Musify.Infrastructure.Messaging.Filters
{
public class ProcessTrackingConsumeFilter<TMessage>(IProcessTrackingStore processTrackingStore)
: IFilter<ConsumeContext<TMessage>>
where TMessage : class
{
public async Task Send(ConsumeContext<TMessage> context, IPipe<ConsumeContext<TMessage>> next)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
typeof(TMessage).Name,
context.CorrelationId ?? context.MessageId,
context.ConversationId,
context.MessageId,
context.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
$"{typeof(TMessage).Name}.Consume",
ProcessStepComponentType.Consumer,
0,
context.CancellationToken);
try
{
await next.Send(context);
await processTrackingStore.CompleteStepAsync(processId, stepId, context.CancellationToken);
}
catch (Exception exception)
{
await processTrackingStore.FailStepAsync(
processId,
stepId,
exception.Message,
context.CancellationToken);
throw;
}
}
public void Probe(ProbeContext context)
{
context.CreateFilterScope(nameof(ProcessTrackingConsumeFilter<TMessage>));
}
}
}
Infrastructure/Messaging/MassTransitEventBus.cs
-16
View File
@@ -1,16 +0,0 @@
using MassTransit;
using Musify.Application.Contracts.Infrastructure;
namespace Musify.Infrastructure.Messaging
{
public class MassTransitEventBus(IPublishEndpoint publishEndpoint)
: IEventBus
{
public async Task PublishAsync<T>(T message, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(message, nameof(message));
await publishEndpoint.Publish(message, cancellationToken);
}
}
}
Infrastructure/Messaging/MessagingDependencyInjection.cs
-139
View File
@@ -1,139 +0,0 @@
using MassTransit;
using Microsoft.Extensions.DependencyInjection;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
using Musify.Infrastructure.Messaging.Consumers;
using Musify.Infrastructure.Messaging.Filters;
namespace Musify.Infrastructure.Messaging
{
public static class MessagingDependencyInjection
{
public static IServiceCollection AddMassTransitClient(this IServiceCollection serviceDescriptors)
{
serviceDescriptors.AddMassTransit(options =>
{
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
ConfigureRabbitMqHost(
busFactoryConfigurator,
busRegistrationContext.GetRequiredService<MessagingConfiguration>()));
} );
return serviceDescriptors;
}
public static IServiceCollection AddMassTransitConsumers(this IServiceCollection serviceDescriptors)
{
serviceDescriptors.AddMassTransit(options =>
{
// Register consumers
options.AddConsumer<UpdatePlayListPictureConsumer>();
options.AddConsumer<RemoveFileConsumer>();
options.AddConsumer<TranscodeAudioFromTrackConsumer>();
options.AddConsumer<UpdateTrackPictureConsumer>();
// Register execute activities
options.AddExecuteActivity<ResizePictureActivity, ResizePictureLocalArguments>();
options.AddExecuteActivity<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>();
options.AddExecuteActivity<UpdatePlayListPictureActivity, UpdatePlayListPictureArguments>();
options.AddExecuteActivity<UpdateTrackPictureActivity, UpdateTrackPictureArguments>();
options.AddExecuteActivity<DownloadFileFromBucketActivity, DownloadFileFromBucketArguments>();
options.AddExecuteActivity<TranscodeDashAudioActivity, TranscodeDashAudioArguments>();
options.AddExecuteActivity<TransferFilesToBucket, TransferFilesToBucketArguments>();
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
{
ConfigureRabbitMqHost(
busFactoryConfigurator,
busRegistrationContext.GetRequiredService<MessagingConfiguration>());
busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext);
// Consumer endpoints
busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<UpdatePlayListPictureConsumer>(busRegistrationContext);
} );
busFactoryConfigurator.ReceiveEndpoint(RemoveFileConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<RemoveFileConsumer>(busRegistrationContext);
} );
busFactoryConfigurator.ReceiveEndpoint(TranscodeAudioFromTrackConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<TranscodeAudioFromTrackConsumer>(busRegistrationContext);
} );
busFactoryConfigurator.ReceiveEndpoint(UpdateTrackPictureConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<UpdateTrackPictureConsumer>(busRegistrationContext);
} );
// Activity endpoints
ConfigureExecuteActivityEndpoint<ResizePictureActivity, ResizePictureLocalArguments>(
busFactoryConfigurator,
busRegistrationContext,
ResizePictureActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>(
busFactoryConfigurator,
busRegistrationContext,
RemoveFileFromBucketActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<UpdatePlayListPictureActivity, UpdatePlayListPictureArguments>(
busFactoryConfigurator,
busRegistrationContext,
UpdatePlayListPictureActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<UpdateTrackPictureActivity, UpdateTrackPictureArguments>(
busFactoryConfigurator,
busRegistrationContext,
UpdateTrackPictureActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<DownloadFileFromBucketActivity, DownloadFileFromBucketArguments>(
busFactoryConfigurator,
busRegistrationContext,
DownloadFileFromBucketActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<TranscodeDashAudioActivity, TranscodeDashAudioArguments>(
busFactoryConfigurator,
busRegistrationContext,
TranscodeDashAudioActivity.ExecuteEndpointName);
ConfigureExecuteActivityEndpoint<TransferFilesToBucket, TransferFilesToBucketArguments>(
busFactoryConfigurator,
busRegistrationContext,
TransferFilesToBucket.ExecuteEndpointName);
} );
} );
return serviceDescriptors;
}
static void ConfigureRabbitMqHost(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
MessagingConfiguration massTransitConfiguration)
{
busFactoryConfigurator.Host(massTransitConfiguration.Host, options =>
{
options.Username(massTransitConfiguration.Username);
options.Password(massTransitConfiguration.Password);
} );
}
static void ConfigureExecuteActivityEndpoint<TActivity, TArguments>(
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
IBusRegistrationContext busRegistrationContext,
string endpointName)
where TActivity : class, IExecuteActivity<TArguments>
where TArguments : class
{
busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator =>
{
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
} );
}
}
}
Infrastructure/Messaging/RoutingSlip/ActivityNames.cs
View File
Infrastructure/Messaging/RoutingSlip/Builders/AudioWorkflowRoutingSlipBuilder.cs
View File
Infrastructure/Messaging/RoutingSlip/Builders/PictureWorkflowRoutingSlipBuilder.cs
View File
Infrastructure/Messaging/RoutingSlip/Builders/RemoveFileRoutingSlipBuilder.cs
View File
Infrastructure/Messaging/RoutingSlipVariableNames.cs
View File