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

Reworks

Browse Source
This commit is contained in:
Admin
2026-05-13 02:20:22 +02:00
parent 5761f6da69
commit d0e6eaaee8
27 changed files with 445 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Infrastructure/MassTransit/Activities/Audio/GenerateAudioWorkflowPathsActivity.cs
+32 -21
View File
@@ -1,4 +1,5 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
@@ -6,42 +7,41 @@ using Musify.Infrastructure.MassTransit.Activities.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities
{
public class GenerateAudioWorkflowPathsActivity(
ILogger<GenerateAudioWorkflowPathsActivity> logger,
IProcessTrackingStore processTrackingStore)
internal class GenerateAudioWorkflowPathsActivity(
IDatabase database,
ILogger<GenerateAudioWorkflowPathsActivity> logger)
: IExecuteActivity<GenerateAudioWorkflowPathsArguments>
{
public const string ExecuteEndpointName = "generate-audio-workflow-paths";
public async Task<ExecutionResult> Execute(ExecuteContext<GenerateAudioWorkflowPathsArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(GenerateAudioWorkflowPathsActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
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;
}
try
{
track.AudioTranscodeProcessingStatus = 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);
Directory.CreateDirectory(workingDirectory);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogDebug("Generated audio workflow paths in {WorkingDirectory}",
workingDirectory);
logger.LogDebug("Generated audio workflow paths in {WorkingDirectory}", workingDirectory);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.CompletedWithVariables(new Dictionary<string, object>
{
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
@@ -54,9 +54,20 @@ namespace Musify.Infrastructure.MassTransit.Activities
{
logger.LogError(exception, "Failed to generate audio workflow paths for {SourceKey}",
executeContext.Arguments.SourceKey);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
try
{
track.AudioTranscodeProcessingStatus = 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);
}
throw;
}
}
}
}
}
Infrastructure/MassTransit/Activities/Audio/TranscodeDashAudioActivity.cs
+34 -28
View File
@@ -1,4 +1,5 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
@@ -7,54 +8,59 @@ using Musify.Infrastructure.MassTransit.Activities.Logs;
namespace Musify.Infrastructure.MassTransit.Activities
{
public class TranscodeDashAudioActivity(
internal class TranscodeDashAudioActivity(
IDatabase database,
IAudioTranscoderService audioTranscoder,
ILogger<TranscodeDashAudioActivity> logger,
IProcessTrackingStore processTrackingStore)
ILogger<TranscodeDashAudioActivity> logger)
: IActivity<TranscodeDashAudioArguments, TranscodeDashAudioLog>
{
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>(executeContext.Arguments.SourceFilePathVariable);
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
var workingDirectory = executeContext.GetVariable<string>(executeContext.Arguments.WorkingDirectoryVariable);
ArgumentNullException.ThrowIfNull(workingDirectory, nameof(workingDirectory));
Track track;
try
{
await using (var fileStream = File.OpenRead(sourceFilePath))
{
await audioTranscoder.TranscodeToDashAsync(
fileStream,
workingDirectory,
executeContext.CancellationToken);
}
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;
}
try
{
await using var fileStream = File.OpenRead(sourceFilePath);
await audioTranscoder.TranscodeToDashAsync(
fileStream,
workingDirectory,
executeContext.CancellationToken);
File.Delete(sourceFilePath);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to transcode {SourceFilePath}",
sourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
logger.LogError(exception, "Failed to transcode {SourceFilePath}", sourceFilePath);
try
{
track.AudioTranscodeProcessingStatus = 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);
}
throw;
}
}
Infrastructure/MassTransit/Activities/Audio/UpdateTrackAudioActivity.cs
+56 -45
View File
@@ -1,4 +1,5 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
@@ -7,63 +8,76 @@ using Musify.Infrastructure.MassTransit.Logs;
namespace Musify.Infrastructure.MassTransit.Activities.Audio
{
public class UpdateTrackAudioActivity(
internal class UpdateTrackAudioActivity(
IDatabase database,
ILogger<TranscodeDashAudioActivity> logger,
IProcessTrackingStore processTrackingStore)
ILogger<UpdateTrackAudioActivity> logger)
: IActivity<UpdateTrackAudioArguments, UpdateTrackAudioLog>
{
public const string ExecuteEndpointName = "update-track-audio";
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateTrackAudioArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
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 {TrackId}",
executeContext.Arguments.TrackId);
throw;
}
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(UpdateTrackAudioActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
string? audioFolderKey;
try
{
audioFolderKey = executeContext.GetVariable<string>(executeContext.Arguments.AudioFolderKeyVariable);
logger.LogDebug("Resolved audio folder key: {AudioFolderKey}", audioFolderKey);
}
catch (Exception exception)
{
logger.LogWarning(exception, "Audio folder key variable could not be resolved for track {TrackId}", track.Id);
return executeContext.Completed();
}
if (string.IsNullOrWhiteSpace(audioFolderKey))
{
logger.LogWarning("Audio folder key is empty for track {TrackId}; skipping update", track.Id);
return executeContext.Completed();
}
try
{
var track = await database.Tracks.FindAsync(
[executeContext.Arguments.TrackId],
cancellationToken: executeContext.CancellationToken);
if (track is null)
{
logger.LogWarning("Track {TrackId} not found",
executeContext.Arguments.TrackId);
throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found");
}
var log = new UpdateTrackAudioLog(
track.Id,
track.AudioFolderName);
track.AudioFolderName = Path.GetFileName(executeContext.Arguments.AudioFolderKey);
database.Tracks.Update(track);
track.AudioFolderName = Path.GetFileName(audioFolderKey);
track.AudioTranscodeProcessingStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated track {TrackId} audio",
executeContext.Arguments.TrackId);
logger.LogInformation("Updated track {TrackId} audio", executeContext.Arguments.TrackId);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
return executeContext.Completed(log);
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to update track {TrackId} audio",
executeContext.Arguments.TrackId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
try
{
track.AudioTranscodeProcessingStatus = 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);
}
throw;
}
}
@@ -72,25 +86,22 @@ namespace Musify.Infrastructure.MassTransit.Activities.Audio
{
try
{
var track = await database.Tracks.FindAsync(
[compensateContext.Log.TrackId],
cancellationToken: compensateContext.CancellationToken);
var track = await database.Tracks.SingleOrDefaultAsync(t => t.Id == compensateContext.Log.TrackId, compensateContext.CancellationToken);
if (track is null)
{
return compensateContext.Compensated();
}
throw new InvalidOperationException($"Track with id {compensateContext.Log.TrackId} not found");
track.AudioFolderName = compensateContext.Log.PreviousFolderAudioKey;
database.Tracks.Update(track);
track.AudioFolderName = compensateContext.Log.PreviousAudioFolderName;
track.AudioTranscodeProcessingStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(compensateContext.CancellationToken);
logger.LogInformation("Compensated track {TrackId} audio", compensateContext.Log.TrackId);
return compensateContext.Compensated();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to compensate track {TrackId} audio", compensateContext.Log.TrackId);
logger.LogError(exception, "Failed to compensate track {TrackId} audio",
compensateContext.Log.TrackId);
return compensateContext.Failed(exception);
}
}
Infrastructure/MassTransit/Activities/Files/DownloadFileFromBucketActivity.cs
+2 -2
View File
@@ -5,9 +5,9 @@ using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments;
using Musify.Infrastructure.MassTransit.Activities.Logs;
namespace Musify.Infrastructure.MassTransit.Activities
namespace Musify.Infrastructure.MassTransit.Activities.Files
{
public class DownloadFileFromBucketActivity(
internal class DownloadFileFromBucketActivity(
IStorageService storageHandler,
ILogger<DownloadFileFromBucketActivity> logger,
IProcessTrackingStore processTrackingStore)
Infrastructure/MassTransit/Activities/Files/RemoveFileFromBucketActivity.cs
+3 -22
View File
@@ -1,43 +1,25 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.MassTransit.Activities.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities
namespace Musify.Infrastructure.MassTransit.Activities.Files
{
public class RemoveFileFromBucketActivity(
internal class RemoveFileFromBucketActivity(
IStorageService storageHandler,
ILogger<RemoveFileFromBucketActivity> logger,
IProcessTrackingStore processTrackingStore)
ILogger<RemoveFileFromBucketActivity> logger)
: IExecuteActivity<RemoveFileFromBucketArguments>
{
public const string ExecuteEndpointName = "remove-file-from-bucket";
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
{
await storageHandler.RemoveFileAsync(
executeContext.Arguments.Bucket,
executeContext.Arguments.Key,
executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
@@ -45,7 +27,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
logger.LogError(exception, "Failed to remove {Key} from bucket {Bucket}",
executeContext.Arguments.Key,
executeContext.Arguments.Bucket);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
Infrastructure/MassTransit/Activities/Files/RemoveFolderFromBucketActivity.cs
+42
View File
@@ -0,0 +1,42 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Infrastructure.MassTransit.Arguments;
namespace Musify.Infrastructure.MassTransit.Activities.Files
{
internal class RemoveFolderFromBucketActivity(
IStorageService storageHandler,
ILogger<RemoveFolderFromBucketActivity> logger)
: IExecuteActivity<RemoveFolderFromBucketArguments>
{
public const string ExecuteEndpointName = "remove-folder-from-bucket";
public async Task<ExecutionResult> Execute(ExecuteContext<RemoveFolderFromBucketArguments> executeContext)
{
var folderKey = executeContext.GetVariable<string>(executeContext.Arguments.FolderKeyVariable);
ArgumentException.ThrowIfNullOrEmpty(folderKey, nameof(folderKey));
try
{
await storageHandler.RemoveFolderAsync(
executeContext.Arguments.Bucket,
folderKey,
executeContext.CancellationToken);
logger.LogInformation("Removed folder {FolderKey} from bucket {Bucket}",
folderKey,
executeContext.Arguments.Bucket);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to remove folder {FolderKey} from bucket {Bucket}",
folderKey,
executeContext.Arguments.Bucket);
throw;
}
}
}
}
Infrastructure/MassTransit/Activities/Pictures/UpdateTrackPictureActivity.cs
+1 -18
View File
@@ -9,28 +9,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
{
public class UpdateTrackPictureActivity(
IDatabase database,
ILogger<UpdateTrackPictureActivity> logger,
IProcessTrackingStore processTrackingStore)
ILogger<UpdateTrackPictureActivity> logger)
: IActivity<UpdateTrackPictureArguments, UpdateTrackPictureLog>
{
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);
var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable);
ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable));
var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable);
@@ -69,14 +54,12 @@ namespace Musify.Infrastructure.MassTransit.Activities
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, "Failed to update track {TrackId} pictures",
executeContext.Arguments.TrackId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
Infrastructure/MassTransit/Activities/Tracks/DeleteTrackActivity.cs
+115
View File
@@ -0,0 +1,115 @@
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Configuration;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
namespace Musify.Infrastructure.MassTransit.Activities.Tracks
{
internal class DeleteTrackActivity(
IDatabase database,
IStorageService storageService,
TrackConfiguration trackConfiguration,
ApplicationStorageConfiguration storageConfiguration,
ILogger<DeleteTrackActivity> logger)
: IExecuteActivity<DeleteTrackArguments>
{
public const string ExecuteEndpointName = "delete-track";
public async Task<ExecutionResult> Execute(ExecuteContext<DeleteTrackArguments> executeContext)
{
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 {TrackId} for deletion", executeContext.Arguments.TrackId);
throw;
}
try
{
await RemovePicturesAsync(track, executeContext.CancellationToken);
track.PicturesProcessingStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
await RemoveAudiosAsync(track, executeContext.CancellationToken);
track.AudioTranscodeProcessingStatus = ProcessingStatus.Completed;
await database.SaveChangesAsync(executeContext.CancellationToken);
database.Tracks.Remove(track);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Deleted track {TrackId}", track.Id);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to delete track {TrackId}", executeContext.Arguments.TrackId);
try
{
track.PicturesProcessingStatus = ProcessingStatus.Failed;
track.AudioTranscodeProcessingStatus = ProcessingStatus.Failed;
await database.SaveChangesAsync(executeContext.CancellationToken);
}
catch (Exception dbException)
{
logger.LogError(dbException, "Failed to update processing statuses to Failed for track {TrackId}", executeContext.Arguments.TrackId);
}
throw;
}
}
async Task RemovePicturesAsync(Track track, CancellationToken cancellationToken)
{
await RemoveFileAsync(trackConfiguration.Routes.BuildOriginalPicturePath(track.OriginalPictureName), cancellationToken);
if (!track.IsPicturesProcessed)
{
logger.LogInformation(
"Track {TrackId} pictures are not fully processed, skipping removal of small, medium, and large pictures",
track.Id);
return;
}
await RemoveFileAsync(trackConfiguration.Routes.BuildSmallPicturePath(track.SmallPictureName), cancellationToken);
await RemoveFileAsync(trackConfiguration.Routes.BuildMediumPicturePath(track.MediumPictureName), cancellationToken);
await RemoveFileAsync(trackConfiguration.Routes.BuildLargePicturePath(track.LargePictureName), cancellationToken);
}
async Task RemoveAudiosAsync(Track track, CancellationToken cancellationToken)
{
await RemoveFileAsync(trackConfiguration.Routes.BuildOriginalAudioPath(track.OriginalAudioName), cancellationToken);
if (!track.IsAudioProcessed)
{
logger.LogInformation(
"Track {TrackId} audio is not fully processed, skipping removal of processed audio",
track.Id);
return;
}
await RemoveFileAsync(trackConfiguration.Routes.BuildProcessedAudioPath(track.AudioFolderName), cancellationToken);
}
async Task RemoveFileAsync(string key, CancellationToken cancellationToken)
{
try
{
logger.LogDebug("Removing file {Key} from bucket {Bucket}", key, storageConfiguration.Bucket);
await storageService.RemoveFileAsync(storageConfiguration.Bucket, key, cancellationToken);
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to remove file {Key} from bucket {Bucket}", key, storageConfiguration.Bucket);
throw;
}
}
}
}
Infrastructure/MassTransit/Activities/Tracks/DeleteTrackArguments.cs
+6
View File
@@ -0,0 +1,6 @@
namespace Musify.Infrastructure.MassTransit.Activities.Tracks
{
internal record DeleteTrackArguments(
Guid TrackId,
Guid UserId);
}