Change editor rule
This commit is contained in:
+20
-19
@@ -3,29 +3,30 @@ using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Events;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Albums;
|
||||
|
||||
internal class PublishAlbumPictureProcessingEventActivity(
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<PublishAlbumPictureProcessingEventActivity> logger)
|
||||
: IExecuteActivity<PublishAlbumPictureProcessingEventArguments>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Albums
|
||||
{
|
||||
public const string ExecuteEndpointName = "publish-album-picture-processing-event";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<PublishAlbumPictureProcessingEventArguments> executeContext)
|
||||
internal class PublishAlbumPictureProcessingEventActivity(
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<PublishAlbumPictureProcessingEventActivity> logger)
|
||||
: IExecuteActivity<PublishAlbumPictureProcessingEventArguments>
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
public const string ExecuteEndpointName = "publish-album-picture-processing-event";
|
||||
|
||||
await publishEndpoint.Publish(
|
||||
new UpdateAlbumPictureEvent(
|
||||
arguments.AlbumId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalPictureKey,
|
||||
arguments.Sizes),
|
||||
executeContext.CancellationToken);
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<PublishAlbumPictureProcessingEventArguments> executeContext)
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
|
||||
logger.LogInformation("Published picture processing event for album {AlbumId}", arguments.AlbumId);
|
||||
await publishEndpoint.Publish(
|
||||
new UpdateAlbumPictureEvent(
|
||||
arguments.AlbumId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalPictureKey,
|
||||
arguments.Sizes),
|
||||
executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed();
|
||||
logger.LogInformation("Published picture processing event for album {AlbumId}", arguments.AlbumId);
|
||||
|
||||
return executeContext.Completed();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+32
-31
@@ -5,43 +5,44 @@ using Musify.Domain.ValueObjects;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.RoutingSlip;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio;
|
||||
|
||||
internal class GenerateAudioWorkflowPathsActivity(
|
||||
IDatabase database,
|
||||
ILogger<GenerateAudioWorkflowPathsActivity> logger)
|
||||
: IExecuteActivity<GenerateAudioWorkflowPathsArguments>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio
|
||||
{
|
||||
public const string ExecuteEndpointName = "generate-audio-workflow-paths";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<GenerateAudioWorkflowPathsArguments> executeContext)
|
||||
internal class GenerateAudioWorkflowPathsActivity(
|
||||
IDatabase database,
|
||||
ILogger<GenerateAudioWorkflowPathsActivity> logger)
|
||||
: IExecuteActivity<GenerateAudioWorkflowPathsArguments>
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
var track = await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
public const string ExecuteEndpointName = "generate-audio-workflow-paths";
|
||||
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<GenerateAudioWorkflowPathsArguments> executeContext)
|
||||
{
|
||||
track.Audio.TranscodeStatus = ProcessingStatus.Processing;
|
||||
var arguments = executeContext.Arguments;
|
||||
var track = await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
|
||||
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);
|
||||
|
||||
logger.LogDebug("Generated audio workflow paths in {WorkingDirectory}", workingDirectory);
|
||||
|
||||
return executeContext.CompletedWithVariables(new Dictionary<string, object>
|
||||
try
|
||||
{
|
||||
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
|
||||
[RoutingSlipVariableNames.Audio.SourceFilePath] = sourceFilePath,
|
||||
[RoutingSlipVariableNames.Audio.TranscodedDirectory] = workingDirectory
|
||||
});
|
||||
}
|
||||
catch
|
||||
{
|
||||
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
throw;
|
||||
track.Audio.TranscodeStatus = ProcessingStatus.Processing;
|
||||
|
||||
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);
|
||||
|
||||
logger.LogDebug("Generated audio workflow paths in {WorkingDirectory}", workingDirectory);
|
||||
|
||||
return executeContext.CompletedWithVariables(new Dictionary<string, object>
|
||||
{
|
||||
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
|
||||
[RoutingSlipVariableNames.Audio.SourceFilePath] = sourceFilePath,
|
||||
[RoutingSlipVariableNames.Audio.TranscodedDirectory] = workingDirectory
|
||||
});
|
||||
}
|
||||
catch
|
||||
{
|
||||
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+15
-14
@@ -3,22 +3,23 @@ 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
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio
|
||||
{
|
||||
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)
|
||||
/// <summary>Shared bookkeeping of the audio processing activities.</summary>
|
||||
internal static class TrackAudioStatus
|
||||
{
|
||||
var track = await database.Tracks.SingleOrDefaultAsync(t => t.Id == trackId, cancellationToken);
|
||||
if (track == null)
|
||||
return;
|
||||
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");
|
||||
|
||||
track.Audio.TranscodeStatus = ProcessingStatus.Failed;
|
||||
await database.SaveChangesAsync(cancellationToken);
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+47
-46
@@ -5,61 +5,62 @@ using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
using Musify.Infrastructure.MassTransit.RoutingSlip;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio;
|
||||
|
||||
internal class TranscodeAudioActivity(
|
||||
IDatabase database,
|
||||
IAudioTranscoderService audioTranscoder,
|
||||
ILogger<TranscodeAudioActivity> logger)
|
||||
: IActivity<TranscodeAudioArguments, TranscodeAudioLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio
|
||||
{
|
||||
public const string ExecuteEndpointName = "transcode-audio";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<TranscodeAudioArguments> executeContext)
|
||||
internal class TranscodeAudioActivity(
|
||||
IDatabase database,
|
||||
IAudioTranscoderService audioTranscoder,
|
||||
ILogger<TranscodeAudioActivity> logger)
|
||||
: IActivity<TranscodeAudioArguments, TranscodeAudioLog>
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
var sourceFilePath = executeContext.GetVariable<string>(arguments.SourceFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath);
|
||||
var workingDirectory = executeContext.GetVariable<string>(arguments.WorkingDirectoryVariable);
|
||||
ArgumentNullException.ThrowIfNull(workingDirectory);
|
||||
public const string ExecuteEndpointName = "transcode-audio";
|
||||
|
||||
await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<TranscodeAudioArguments> executeContext)
|
||||
{
|
||||
AudioTranscodeResult transcodeResult;
|
||||
await using (var fileStream = File.OpenRead(sourceFilePath))
|
||||
var arguments = executeContext.Arguments;
|
||||
var sourceFilePath = executeContext.GetVariable<string>(arguments.SourceFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath);
|
||||
var workingDirectory = executeContext.GetVariable<string>(arguments.WorkingDirectoryVariable);
|
||||
ArgumentNullException.ThrowIfNull(workingDirectory);
|
||||
|
||||
await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
transcodeResult = await audioTranscoder.TranscodeToAudioFileAsync(
|
||||
fileStream, workingDirectory, executeContext.CancellationToken);
|
||||
AudioTranscodeResult transcodeResult;
|
||||
await using (var fileStream = File.OpenRead(sourceFilePath))
|
||||
{
|
||||
transcodeResult = await audioTranscoder.TranscodeToAudioFileAsync(
|
||||
fileStream, workingDirectory, executeContext.CancellationToken);
|
||||
}
|
||||
|
||||
if (transcodeResult.StatusCode != 0)
|
||||
throw new InvalidOperationException($"ffmpeg transcoding failed with exit code {transcodeResult.StatusCode}.");
|
||||
|
||||
File.Delete(sourceFilePath);
|
||||
|
||||
return executeContext.CompletedWithVariables(new TranscodeAudioLog(workingDirectory), new Dictionary<string, object>
|
||||
{
|
||||
[RoutingSlipVariableNames.Audio.DurationSeconds] = transcodeResult.Duration.TotalSeconds
|
||||
});
|
||||
}
|
||||
|
||||
if (transcodeResult.StatusCode != 0)
|
||||
throw new InvalidOperationException($"ffmpeg transcoding failed with exit code {transcodeResult.StatusCode}.");
|
||||
|
||||
File.Delete(sourceFilePath);
|
||||
|
||||
return executeContext.CompletedWithVariables(new TranscodeAudioLog(workingDirectory), new Dictionary<string, object>
|
||||
catch
|
||||
{
|
||||
[RoutingSlipVariableNames.Audio.DurationSeconds] = transcodeResult.Duration.TotalSeconds
|
||||
});
|
||||
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
catch
|
||||
|
||||
public Task<CompensationResult> Compensate(CompensateContext<TranscodeAudioLog> compensateContext)
|
||||
{
|
||||
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
throw;
|
||||
var workingDirectory = compensateContext.Log.WorkingDirectory;
|
||||
|
||||
if (Directory.Exists(workingDirectory))
|
||||
Directory.Delete(workingDirectory, recursive: true);
|
||||
else
|
||||
logger.LogWarning("Working directory {WorkingDirectory} not found during compensation", workingDirectory);
|
||||
|
||||
return Task.FromResult(compensateContext.Compensated());
|
||||
}
|
||||
}
|
||||
|
||||
public Task<CompensationResult> Compensate(CompensateContext<TranscodeAudioLog> compensateContext)
|
||||
{
|
||||
var workingDirectory = compensateContext.Log.WorkingDirectory;
|
||||
|
||||
if (Directory.Exists(workingDirectory))
|
||||
Directory.Delete(workingDirectory, recursive: true);
|
||||
else
|
||||
logger.LogWarning("Working directory {WorkingDirectory} not found during compensation", workingDirectory);
|
||||
|
||||
return Task.FromResult(compensateContext.Compensated());
|
||||
}
|
||||
}
|
||||
|
||||
+44
-43
@@ -6,61 +6,62 @@ using Musify.Domain.ValueObjects;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio;
|
||||
|
||||
internal class UpdateTrackAudioActivity(
|
||||
IDatabase database,
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<UpdateTrackAudioActivity> logger)
|
||||
: IActivity<UpdateTrackAudioArguments, UpdateTrackAudioLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Audio
|
||||
{
|
||||
public const string ExecuteEndpointName = "update-track-audio";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateTrackAudioArguments> executeContext)
|
||||
internal class UpdateTrackAudioActivity(
|
||||
IDatabase database,
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<UpdateTrackAudioActivity> logger)
|
||||
: IActivity<UpdateTrackAudioArguments, UpdateTrackAudioLog>
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
var track = await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
public const string ExecuteEndpointName = "update-track-audio";
|
||||
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UpdateTrackAudioArguments> executeContext)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(arguments.AudioFolderKey))
|
||||
throw new InvalidOperationException($"Audio folder key is empty for track {track.Id}");
|
||||
var arguments = executeContext.Arguments;
|
||||
var track = await TrackAudioStatus.LoadAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
|
||||
var durationSeconds = executeContext.GetVariable<double>(arguments.DurationSecondsVariable)
|
||||
?? throw new InvalidOperationException($"Missing routing slip variable {arguments.DurationSecondsVariable}");
|
||||
try
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(arguments.AudioFolderKey))
|
||||
throw new InvalidOperationException($"Audio folder key is empty for track {track.Id}");
|
||||
|
||||
var log = new UpdateTrackAudioLog(track.Id, track.Audio.FolderName, track.DurationSeconds);
|
||||
var durationSeconds = executeContext.GetVariable<double>(arguments.DurationSecondsVariable)
|
||||
?? throw new InvalidOperationException($"Missing routing slip variable {arguments.DurationSecondsVariable}");
|
||||
|
||||
track.Audio.FolderName = Path.GetFileName(arguments.AudioFolderKey);
|
||||
track.DurationSeconds = durationSeconds;
|
||||
track.Audio.TranscodeStatus = ProcessingStatus.Completed;
|
||||
await database.SaveChangesAsync(executeContext.CancellationToken);
|
||||
var log = new UpdateTrackAudioLog(track.Id, track.Audio.FolderName, track.DurationSeconds);
|
||||
|
||||
logger.LogInformation("Updated track {TrackId} audio", arguments.TrackId);
|
||||
track.Audio.FolderName = Path.GetFileName(arguments.AudioFolderKey);
|
||||
track.DurationSeconds = durationSeconds;
|
||||
track.Audio.TranscodeStatus = ProcessingStatus.Completed;
|
||||
await database.SaveChangesAsync(executeContext.CancellationToken);
|
||||
|
||||
await publishEndpoint.Publish(new TrackAudioProcessed(arguments.TrackId), executeContext.CancellationToken);
|
||||
logger.LogInformation("Updated track {TrackId} audio", arguments.TrackId);
|
||||
|
||||
return executeContext.Completed(log);
|
||||
await publishEndpoint.Publish(new TrackAudioProcessed(arguments.TrackId), executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed(log);
|
||||
}
|
||||
catch
|
||||
{
|
||||
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
catch
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<UpdateTrackAudioLog> compensateContext)
|
||||
{
|
||||
await TrackAudioStatus.MarkFailedAsync(database, arguments.TrackId, executeContext.CancellationToken);
|
||||
throw;
|
||||
var log = compensateContext.Log;
|
||||
|
||||
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", log.TrackId);
|
||||
|
||||
return compensateContext.Compensated();
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<UpdateTrackAudioLog> compensateContext)
|
||||
{
|
||||
var log = compensateContext.Log;
|
||||
|
||||
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", log.TrackId);
|
||||
|
||||
return compensateContext.Compensated();
|
||||
}
|
||||
}
|
||||
|
||||
+34
-33
@@ -4,45 +4,46 @@ using Musify.Application.Contracts;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files;
|
||||
|
||||
internal class CopyFileInBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<CopyFileInBucketActivity> logger)
|
||||
: IActivity<CopyFileInBucketArguments, CopyFileInBucketLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
public const string ExecuteEndpointName = "copy-file-in-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<CopyFileInBucketArguments> executeContext)
|
||||
internal class CopyFileInBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<CopyFileInBucketActivity> logger)
|
||||
: IActivity<CopyFileInBucketArguments, CopyFileInBucketLog>
|
||||
{
|
||||
await storageService.CopyFileAsync(
|
||||
executeContext.Arguments.SourceBucket,
|
||||
executeContext.Arguments.SourceKey,
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
executeContext.Arguments.DestinationKey,
|
||||
executeContext.CancellationToken);
|
||||
public const string ExecuteEndpointName = "copy-file-in-bucket";
|
||||
|
||||
return executeContext.Completed(new CopyFileInBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
executeContext.Arguments.DestinationKey));
|
||||
}
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<CopyFileInBucketLog> compensateContext)
|
||||
{
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<CopyFileInBucketArguments> executeContext)
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey,
|
||||
compensateContext.CancellationToken);
|
||||
return compensateContext.Compensated();
|
||||
await storageService.CopyFileAsync(
|
||||
executeContext.Arguments.SourceBucket,
|
||||
executeContext.Arguments.SourceKey,
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
executeContext.Arguments.DestinationKey,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed(new CopyFileInBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
executeContext.Arguments.DestinationKey));
|
||||
}
|
||||
catch (Exception exception)
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<CopyFileInBucketLog> compensateContext)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate copied file {Bucket}/{Key}",
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey);
|
||||
return compensateContext.Failed(exception);
|
||||
try
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey,
|
||||
compensateContext.CancellationToken);
|
||||
return compensateContext.Compensated();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate copied file {Bucket}/{Key}",
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey);
|
||||
return compensateContext.Failed(exception);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+41
-40
@@ -4,52 +4,53 @@ using Musify.Application.Contracts;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files;
|
||||
|
||||
internal class DownloadFileFromBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<DownloadFileFromBucketActivity> logger)
|
||||
: IActivity<DownloadFileFromBucketArguments, DownloadFileFromBucketLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
public const string ExecuteEndpointName = "download-file-from-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext)
|
||||
internal class DownloadFileFromBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<DownloadFileFromBucketActivity> logger)
|
||||
: IActivity<DownloadFileFromBucketArguments, DownloadFileFromBucketLog>
|
||||
{
|
||||
var destinationPath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(destinationPath);
|
||||
public const string ExecuteEndpointName = "download-file-from-bucket";
|
||||
|
||||
using var fileStream = await storageService.GetFileAsync(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
executeContext.CancellationToken)
|
||||
?? throw new FileNotFoundException($"Object {executeContext.Arguments.Bucket}/{executeContext.Arguments.Key} was not found.");
|
||||
|
||||
var destinationDirectory = Path.GetDirectoryName(destinationPath);
|
||||
if (!string.IsNullOrWhiteSpace(destinationDirectory))
|
||||
Directory.CreateDirectory(destinationDirectory);
|
||||
|
||||
if (fileStream.CanSeek)
|
||||
fileStream.Position = 0;
|
||||
|
||||
using var destinationStream = File.Create(destinationPath);
|
||||
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
|
||||
return executeContext.Completed(new DownloadFileFromBucketLog(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
destinationPath));
|
||||
}
|
||||
|
||||
public Task<CompensationResult> Compensate(CompensateContext<DownloadFileFromBucketLog> compensateContext)
|
||||
{
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext)
|
||||
{
|
||||
File.Delete(compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Compensated());
|
||||
var destinationPath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(destinationPath);
|
||||
|
||||
using var fileStream = await storageService.GetFileAsync(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
executeContext.CancellationToken)
|
||||
?? throw new FileNotFoundException($"Object {executeContext.Arguments.Bucket}/{executeContext.Arguments.Key} was not found.");
|
||||
|
||||
var destinationDirectory = Path.GetDirectoryName(destinationPath);
|
||||
if (!string.IsNullOrWhiteSpace(destinationDirectory))
|
||||
Directory.CreateDirectory(destinationDirectory);
|
||||
|
||||
if (fileStream.CanSeek)
|
||||
fileStream.Position = 0;
|
||||
|
||||
using var destinationStream = File.Create(destinationPath);
|
||||
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
|
||||
return executeContext.Completed(new DownloadFileFromBucketLog(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
destinationPath));
|
||||
}
|
||||
catch (Exception exception)
|
||||
|
||||
public Task<CompensationResult> Compensate(CompensateContext<DownloadFileFromBucketLog> compensateContext)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate downloaded file {DestinationFilePath}", compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Failed(exception));
|
||||
try
|
||||
{
|
||||
File.Delete(compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Compensated());
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate downloaded file {DestinationFilePath}", compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Failed(exception));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+26
-25
@@ -5,34 +5,35 @@ using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files;
|
||||
|
||||
internal class RemoveFileFromBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<RemoveFileFromBucketActivity> logger)
|
||||
: IExecuteActivity<RemoveFileFromBucketArguments>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
public const string ExecuteEndpointName = "remove-file-from-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<RemoveFileFromBucketArguments> executeContext)
|
||||
internal class RemoveFileFromBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<RemoveFileFromBucketActivity> logger)
|
||||
: IExecuteActivity<RemoveFileFromBucketArguments>
|
||||
{
|
||||
try
|
||||
public const string ExecuteEndpointName = "remove-file-from-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<RemoveFileFromBucketArguments> executeContext)
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
executeContext.CancellationToken);
|
||||
return executeContext.Completed();
|
||||
}
|
||||
catch (AmazonS3Exception exception) when (
|
||||
exception.StatusCode == HttpStatusCode.NotFound
|
||||
|| string.Equals(exception.ErrorCode, "NoSuchKey", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
logger.LogInformation(
|
||||
"Key {Key} not found in bucket {Bucket}, skipping",
|
||||
executeContext.Arguments.Key,
|
||||
executeContext.Arguments.Bucket);
|
||||
return executeContext.Completed();
|
||||
try
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
executeContext.CancellationToken);
|
||||
return executeContext.Completed();
|
||||
}
|
||||
catch (AmazonS3Exception exception) when (
|
||||
exception.StatusCode == HttpStatusCode.NotFound
|
||||
|| string.Equals(exception.ErrorCode, "NoSuchKey", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
logger.LogInformation(
|
||||
"Key {Key} not found in bucket {Bucket}, skipping",
|
||||
executeContext.Arguments.Key,
|
||||
executeContext.Arguments.Bucket);
|
||||
return executeContext.Completed();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+39
-38
@@ -4,51 +4,52 @@ using Musify.Application.Contracts;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files;
|
||||
|
||||
internal class TransferFilesToBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<TransferFilesToBucketActivity> logger)
|
||||
: IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
public const string ExecuteEndpointName = "transfer-files-to-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext)
|
||||
internal class TransferFilesToBucketActivity(
|
||||
IStorageService storageService,
|
||||
ILogger<TransferFilesToBucketActivity> logger)
|
||||
: IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog>
|
||||
{
|
||||
var folderPath = executeContext.GetVariable<string>(executeContext.Arguments.SourceDirectoryVariable);
|
||||
ArgumentNullException.ThrowIfNull(folderPath);
|
||||
var destinationKey = executeContext.Arguments.DestinationKey;
|
||||
ArgumentNullException.ThrowIfNull(destinationKey);
|
||||
public const string ExecuteEndpointName = "transfer-files-to-bucket";
|
||||
|
||||
var uploadedKeys = await storageService.TransferFilesAsync(
|
||||
folderPath,
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed(new TransferFilesToBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
uploadedKeys.ToArray()));
|
||||
}
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<TransferFilesToBucketLog> compensateContext)
|
||||
{
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext)
|
||||
{
|
||||
foreach (var uploadedKey in compensateContext.Log.UploadedKeys)
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
compensateContext.Log.DestinationBucket,
|
||||
uploadedKey,
|
||||
compensateContext.CancellationToken);
|
||||
}
|
||||
var folderPath = executeContext.GetVariable<string>(executeContext.Arguments.SourceDirectoryVariable);
|
||||
ArgumentNullException.ThrowIfNull(folderPath);
|
||||
var destinationKey = executeContext.Arguments.DestinationKey;
|
||||
ArgumentNullException.ThrowIfNull(destinationKey);
|
||||
|
||||
return compensateContext.Compensated();
|
||||
var uploadedKeys = await storageService.TransferFilesAsync(
|
||||
folderPath,
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed(new TransferFilesToBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
uploadedKeys.ToArray()));
|
||||
}
|
||||
catch (Exception exception)
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<TransferFilesToBucketLog> compensateContext)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate file transfers to {DestinationBucket}", compensateContext.Log.DestinationBucket);
|
||||
return compensateContext.Failed(exception);
|
||||
try
|
||||
{
|
||||
foreach (var uploadedKey in compensateContext.Log.UploadedKeys)
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
compensateContext.Log.DestinationBucket,
|
||||
uploadedKey,
|
||||
compensateContext.CancellationToken);
|
||||
}
|
||||
|
||||
return compensateContext.Compensated();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate file transfers to {DestinationBucket}", compensateContext.Log.DestinationBucket);
|
||||
return compensateContext.Failed(exception);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+38
-37
@@ -4,49 +4,50 @@ using Musify.Application.Contracts;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files;
|
||||
|
||||
internal class UploadFileToBucketActivity(IStorageService storageService,
|
||||
ILogger<UploadFileToBucketActivity> logger)
|
||||
: IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
public const string ExecuteEndpointName = "upload-file-to-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> executeContext)
|
||||
internal class UploadFileToBucketActivity(IStorageService storageService,
|
||||
ILogger<UploadFileToBucketActivity> logger)
|
||||
: IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
|
||||
{
|
||||
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath);
|
||||
var fileName = Path.GetFileName(sourceFilePath);
|
||||
var destinationKey = string.Join('/', new[] { executeContext.Arguments.DestinationRoute, fileName }
|
||||
.Where(static s => !string.IsNullOrWhiteSpace(s))
|
||||
.Select(static s => s.Trim().Trim('/', '\\')));
|
||||
public const string ExecuteEndpointName = "upload-file-to-bucket";
|
||||
|
||||
await storageService.UploadFileAsync(
|
||||
sourceFilePath,
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey,
|
||||
executeContext.CancellationToken);
|
||||
return executeContext.CompletedWithVariables(new UploadFileToBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey));
|
||||
}
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<UploadFileToBucketLog> compensateContext)
|
||||
{
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> executeContext)
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey,
|
||||
compensateContext.CancellationToken);
|
||||
return compensateContext.Compensated();
|
||||
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath);
|
||||
var fileName = Path.GetFileName(sourceFilePath);
|
||||
var destinationKey = string.Join('/', new[] { executeContext.Arguments.DestinationRoute, fileName }
|
||||
.Where(static s => !string.IsNullOrWhiteSpace(s))
|
||||
.Select(static s => s.Trim().Trim('/', '\\')));
|
||||
|
||||
await storageService.UploadFileAsync(
|
||||
sourceFilePath,
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey,
|
||||
executeContext.CancellationToken);
|
||||
return executeContext.CompletedWithVariables(new UploadFileToBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey));
|
||||
}
|
||||
catch (Exception exception)
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<UploadFileToBucketLog> compensateContext)
|
||||
{
|
||||
logger.LogError(exception, "Error compensating file upload for {Bucket}/{Key}",
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey);
|
||||
return compensateContext.Failed(exception);
|
||||
try
|
||||
{
|
||||
await storageService.RemoveFileAsync(
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey,
|
||||
compensateContext.CancellationToken);
|
||||
return compensateContext.Compensated();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Error compensating file upload for {Bucket}/{Key}",
|
||||
compensateContext.Log.DestinationBucket,
|
||||
compensateContext.Log.DestinationKey);
|
||||
return compensateContext.Failed(exception);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+75
-74
@@ -6,92 +6,93 @@ 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
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.LifeCycle
|
||||
{
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<MarkLifeCycleArguments> executeContext)
|
||||
/// <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
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
var entity = await database.Set<TEntity>()
|
||||
.SingleOrDefaultAsync(e => e.Id == arguments.Id, executeContext.CancellationToken);
|
||||
|
||||
if (entity == null)
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<MarkLifeCycleArguments> executeContext)
|
||||
{
|
||||
logger.LogInformation("{Entity} {Id} not found while marking it as {Status}, skipping", typeof(TEntity).Name, arguments.Id, arguments.Status);
|
||||
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();
|
||||
}
|
||||
|
||||
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)
|
||||
/// <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
|
||||
{
|
||||
var id = executeContext.Arguments.Id;
|
||||
var entity = await database.Set<TEntity>()
|
||||
.SingleOrDefaultAsync(e => e.Id == id, executeContext.CancellationToken);
|
||||
|
||||
if (entity == null)
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<DeleteEntityArguments> executeContext)
|
||||
{
|
||||
logger.LogInformation("{Entity} {Id} not found while deleting it, skipping", typeof(TEntity).Name, id);
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
database.Set<TEntity>().Remove(entity);
|
||||
await database.SaveChangesAsync(executeContext.CancellationToken);
|
||||
// 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";
|
||||
}
|
||||
|
||||
logger.LogInformation("Deleted {Entity} {Id} from the database", typeof(TEntity).Name, id);
|
||||
return executeContext.Completed();
|
||||
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";
|
||||
}
|
||||
}
|
||||
|
||||
// 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";
|
||||
}
|
||||
|
||||
+28
-27
@@ -3,37 +3,38 @@ using Microsoft.Extensions.Logging;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.RoutingSlip;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
|
||||
|
||||
internal class GeneratePictureWorkflowPathsActivity(
|
||||
ILogger<GeneratePictureWorkflowPathsActivity> logger)
|
||||
: IExecuteActivity<GeneratePictureWorkflowPathsArguments>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Pictures
|
||||
{
|
||||
public const string ExecuteEndpointName = "generate-picture-workflow-paths";
|
||||
|
||||
internal const string ResizedPictureFileExtension = ".webp";
|
||||
|
||||
public Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext)
|
||||
internal class GeneratePictureWorkflowPathsActivity(
|
||||
ILogger<GeneratePictureWorkflowPathsActivity> logger)
|
||||
: IExecuteActivity<GeneratePictureWorkflowPathsArguments>
|
||||
{
|
||||
var workingDirectory = Path.Combine(executeContext.Arguments.TemporaryRootDirectory, Guid.NewGuid().ToString());
|
||||
public const string ExecuteEndpointName = "generate-picture-workflow-paths";
|
||||
|
||||
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);
|
||||
internal const string ResizedPictureFileExtension = ".webp";
|
||||
|
||||
Directory.CreateDirectory(workingDirectory);
|
||||
|
||||
logger.LogDebug("Generated picture workflow paths in {WorkingDirectory}", workingDirectory);
|
||||
|
||||
return Task.FromResult(executeContext.CompletedWithVariables(new Dictionary<string, object>
|
||||
public Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext)
|
||||
{
|
||||
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
|
||||
[RoutingSlipVariableNames.Picture.OriginalFilePath] = sourceFilePath,
|
||||
[RoutingSlipVariableNames.Picture.SmallResizedFilePath] = smallPictureFilePath,
|
||||
[RoutingSlipVariableNames.Picture.MediumResizedFilePath] = mediumPictureFilePath,
|
||||
[RoutingSlipVariableNames.Picture.LargeResizedFilePath] = largePictureFilePath
|
||||
}));
|
||||
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);
|
||||
|
||||
Directory.CreateDirectory(workingDirectory);
|
||||
|
||||
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
|
||||
}));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+48
-47
@@ -4,60 +4,61 @@ using Musify.Application.Contracts;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Pictures;
|
||||
|
||||
internal class ResizePictureActivity(
|
||||
IPictureService pictureHandler,
|
||||
ILogger<ResizePictureActivity> logger)
|
||||
: IActivity<ResizePictureLocalArguments, ResizePictureLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Pictures
|
||||
{
|
||||
public const string ExecuteEndpointName = "resize-picture";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
|
||||
internal class ResizePictureActivity(
|
||||
IPictureService pictureHandler,
|
||||
ILogger<ResizePictureActivity> logger)
|
||||
: IActivity<ResizePictureLocalArguments, ResizePictureLog>
|
||||
{
|
||||
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath);
|
||||
var destinationFilePath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(destinationFilePath);
|
||||
public const string ExecuteEndpointName = "resize-picture";
|
||||
|
||||
if (!File.Exists(sourceFilePath))
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
|
||||
{
|
||||
logger.LogWarning("Source file not found: {SourceFilePath}", sourceFilePath);
|
||||
throw new FileNotFoundException($"Source file not found: {sourceFilePath}");
|
||||
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath);
|
||||
var destinationFilePath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(destinationFilePath);
|
||||
|
||||
if (!File.Exists(sourceFilePath))
|
||||
{
|
||||
logger.LogWarning("Source file not found: {SourceFilePath}", sourceFilePath);
|
||||
throw new FileNotFoundException($"Source file not found: {sourceFilePath}");
|
||||
}
|
||||
|
||||
await using var fileStream = File.OpenRead(sourceFilePath);
|
||||
|
||||
await using var resizedPicture = await pictureHandler.ResizePictureAsWebpAsync(
|
||||
fileStream,
|
||||
executeContext.Arguments.Width,
|
||||
executeContext.Arguments.Height,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var destinationDirectory = Path.GetDirectoryName(destinationFilePath);
|
||||
if (!string.IsNullOrEmpty(destinationDirectory))
|
||||
Directory.CreateDirectory(destinationDirectory);
|
||||
|
||||
await using var destinationStream = File.Create(destinationFilePath);
|
||||
await resizedPicture.CopyToAsync(destinationStream, executeContext.CancellationToken);
|
||||
|
||||
logger.LogDebug("Resized picture from {SourceFilePath} to {DestinationFilePath}",
|
||||
sourceFilePath,
|
||||
destinationFilePath);
|
||||
return executeContext.Completed(new ResizePictureLog(destinationFilePath));
|
||||
}
|
||||
|
||||
await using var fileStream = File.OpenRead(sourceFilePath);
|
||||
|
||||
await using var resizedPicture = await pictureHandler.ResizePictureAsWebpAsync(
|
||||
fileStream,
|
||||
executeContext.Arguments.Width,
|
||||
executeContext.Arguments.Height,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var destinationDirectory = Path.GetDirectoryName(destinationFilePath);
|
||||
if (!string.IsNullOrEmpty(destinationDirectory))
|
||||
Directory.CreateDirectory(destinationDirectory);
|
||||
|
||||
await using var destinationStream = File.Create(destinationFilePath);
|
||||
await resizedPicture.CopyToAsync(destinationStream, executeContext.CancellationToken);
|
||||
|
||||
logger.LogDebug("Resized picture from {SourceFilePath} to {DestinationFilePath}",
|
||||
sourceFilePath,
|
||||
destinationFilePath);
|
||||
return executeContext.Completed(new ResizePictureLog(destinationFilePath));
|
||||
}
|
||||
|
||||
public Task<CompensationResult> Compensate(CompensateContext<ResizePictureLog> compensateContext)
|
||||
{
|
||||
try
|
||||
public Task<CompensationResult> Compensate(CompensateContext<ResizePictureLog> compensateContext)
|
||||
{
|
||||
File.Delete(compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Compensated());
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate resized file {DestinationFilePath}", compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Failed(exception));
|
||||
try
|
||||
{
|
||||
File.Delete(compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Compensated());
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate resized file {DestinationFilePath}", compensateContext.Log.DestinationFilePath);
|
||||
return Task.FromResult(compensateContext.Failed(exception));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+124
-123
@@ -8,153 +8,154 @@ 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
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Pictures
|
||||
{
|
||||
protected IDatabase Database { get; } = database;
|
||||
internal sealed record PictureNames(string? Original, string? Small, string? Medium, string? Large);
|
||||
|
||||
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)
|
||||
/// <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
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
var small = RequireVariable(executeContext, arguments.SmallPictureVariable);
|
||||
var medium = RequireVariable(executeContext, arguments.MediumPictureVariable);
|
||||
var large = RequireVariable(executeContext, arguments.LargePictureVariable);
|
||||
protected IDatabase Database { get; } = database;
|
||||
|
||||
var entity = await FindAsync(arguments.SubjectId, executeContext.CancellationToken)
|
||||
?? throw new InvalidOperationException($"{typeof(TEntity).Name} {arguments.SubjectId} not found");
|
||||
protected abstract Task<TEntity?> FindAsync(Guid id, CancellationToken cancellationToken);
|
||||
|
||||
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);
|
||||
/// <summary>Reads the current names; throws when the entity has no pending picture.</summary>
|
||||
protected abstract PictureNames Read(TEntity entity);
|
||||
|
||||
logger.LogInformation("Updated {Entity} {Id} pictures", typeof(TEntity).Name, arguments.SubjectId);
|
||||
protected abstract void Write(TEntity entity, PictureNames names, bool succeeded);
|
||||
|
||||
await publishEndpoint.Publish(ProcessedEvent(arguments.SubjectId), executeContext.CancellationToken);
|
||||
protected abstract object ProcessedEvent(Guid id);
|
||||
|
||||
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
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePicturesArguments> executeContext)
|
||||
{
|
||||
var entity = await FindAsync(log.SubjectId, compensateContext.CancellationToken);
|
||||
if (entity == null)
|
||||
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();
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
catch (Exception 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)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate {Entity} {Id} pictures", typeof(TEntity).Name, log.SubjectId);
|
||||
return compensateContext.Failed(exception);
|
||||
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}");
|
||||
}
|
||||
|
||||
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)
|
||||
internal sealed class UpdateTrackPictureActivity(IDatabase database, IPublishEndpoint publishEndpoint, ILogger<UpdateTrackPictureActivity> logger)
|
||||
: UpdatePicturesActivity<Track>(database, publishEndpoint, logger)
|
||||
{
|
||||
if (pictures == null)
|
||||
return;
|
||||
public const string ExecuteEndpointName = "update-track-picture";
|
||||
|
||||
pictures.OriginalName = names.Original ?? pictures.OriginalName;
|
||||
pictures.SmallName = names.Small;
|
||||
pictures.MediumName = names.Medium;
|
||||
pictures.LargeName = names.Large;
|
||||
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);
|
||||
}
|
||||
|
||||
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)
|
||||
internal sealed class UpdatePlayListPictureActivity(IDatabase database, IPublishEndpoint publishEndpoint, ILogger<UpdatePlayListPictureActivity> logger)
|
||||
: UpdatePicturesActivity<PlayList>(database, publishEndpoint, logger)
|
||||
{
|
||||
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;
|
||||
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);
|
||||
}
|
||||
|
||||
protected override object ProcessedEvent(Guid id) => new TrackPictureProcessed(id);
|
||||
}
|
||||
internal sealed class UpdateAlbumPictureActivity(IDatabase database, IPublishEndpoint publishEndpoint, ILogger<UpdateAlbumPictureActivity> logger)
|
||||
: UpdatePicturesActivity<Album>(database, publishEndpoint, logger)
|
||||
{
|
||||
public const string ExecuteEndpointName = "update-album-picture";
|
||||
|
||||
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<Album?> FindAsync(Guid id, CancellationToken cancellationToken) =>
|
||||
Database.Albums.SingleOrDefaultAsync(album => album.Id == id, cancellationToken);
|
||||
|
||||
protected override Task<PlayList?> FindAsync(Guid id, CancellationToken cancellationToken) =>
|
||||
Database.PlayLists.SingleOrDefaultAsync(playList => playList.Id == id, cancellationToken);
|
||||
protected override PictureNames Read(Album album) => ReadPictures(album.Pictures);
|
||||
|
||||
protected override PictureNames Read(PlayList playList) => ReadPictures(playList.Pictures);
|
||||
protected override void Write(Album album, PictureNames names, bool succeeded) =>
|
||||
WritePictures(album.Pictures, names);
|
||||
|
||||
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);
|
||||
protected override object ProcessedEvent(Guid id) => new AlbumPictureProcessed(id);
|
||||
}
|
||||
}
|
||||
|
||||
+20
-19
@@ -3,29 +3,30 @@ using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Events;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.PlayLists;
|
||||
|
||||
internal class PublishPlayListPictureProcessingEventActivity(
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<PublishPlayListPictureProcessingEventActivity> logger)
|
||||
: IExecuteActivity<PublishPlayListPictureProcessingEventArguments>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.PlayLists
|
||||
{
|
||||
public const string ExecuteEndpointName = "publish-play-list-picture-processing-event";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<PublishPlayListPictureProcessingEventArguments> executeContext)
|
||||
internal class PublishPlayListPictureProcessingEventActivity(
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<PublishPlayListPictureProcessingEventActivity> logger)
|
||||
: IExecuteActivity<PublishPlayListPictureProcessingEventArguments>
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
public const string ExecuteEndpointName = "publish-play-list-picture-processing-event";
|
||||
|
||||
await publishEndpoint.Publish(
|
||||
new UpdatePlayListPictureEvent(
|
||||
arguments.PlayListId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalPictureKey,
|
||||
arguments.Sizes),
|
||||
executeContext.CancellationToken);
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<PublishPlayListPictureProcessingEventArguments> executeContext)
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
|
||||
logger.LogInformation("Published picture processing event for playlist {PlayListId}", arguments.PlayListId);
|
||||
await publishEndpoint.Publish(
|
||||
new UpdatePlayListPictureEvent(
|
||||
arguments.PlayListId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalPictureKey,
|
||||
arguments.Sizes),
|
||||
executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed();
|
||||
logger.LogInformation("Published picture processing event for playlist {PlayListId}", arguments.PlayListId);
|
||||
|
||||
return executeContext.Completed();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+28
-27
@@ -3,38 +3,39 @@ using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Events;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Tracks;
|
||||
|
||||
internal class PublishTrackProcessingEventsActivity(
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<PublishTrackProcessingEventsActivity> logger)
|
||||
: IExecuteActivity<PublishTrackProcessingEventsArguments>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Tracks
|
||||
{
|
||||
public const string ExecuteEndpointName = "publish-track-processing-events";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<PublishTrackProcessingEventsArguments> executeContext)
|
||||
internal class PublishTrackProcessingEventsActivity(
|
||||
IPublishEndpoint publishEndpoint,
|
||||
ILogger<PublishTrackProcessingEventsActivity> logger)
|
||||
: IExecuteActivity<PublishTrackProcessingEventsArguments>
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
public const string ExecuteEndpointName = "publish-track-processing-events";
|
||||
|
||||
await publishEndpoint.Publish(
|
||||
new UpdateTrackPictureEvent(
|
||||
arguments.TrackId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalPictureKey,
|
||||
arguments.Sizes),
|
||||
executeContext.CancellationToken);
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<PublishTrackProcessingEventsArguments> executeContext)
|
||||
{
|
||||
var arguments = executeContext.Arguments;
|
||||
|
||||
await publishEndpoint.Publish(
|
||||
new UpdateTrackAudioEvent(
|
||||
arguments.TrackId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalAudioKey,
|
||||
arguments.Bucket,
|
||||
arguments.AudioDestinationFolderKey),
|
||||
executeContext.CancellationToken);
|
||||
await publishEndpoint.Publish(
|
||||
new UpdateTrackPictureEvent(
|
||||
arguments.TrackId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalPictureKey,
|
||||
arguments.Sizes),
|
||||
executeContext.CancellationToken);
|
||||
|
||||
logger.LogInformation("Published processing events for track {TrackId}", arguments.TrackId);
|
||||
await publishEndpoint.Publish(
|
||||
new UpdateTrackAudioEvent(
|
||||
arguments.TrackId,
|
||||
arguments.Bucket,
|
||||
arguments.FinalAudioKey,
|
||||
arguments.Bucket,
|
||||
arguments.AudioDestinationFolderKey),
|
||||
executeContext.CancellationToken);
|
||||
|
||||
return executeContext.Completed();
|
||||
logger.LogInformation("Published processing events for track {TrackId}", arguments.TrackId);
|
||||
|
||||
return executeContext.Completed();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+37
-36
@@ -6,56 +6,57 @@ using Musify.Domain.ValueObjects;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.UploadIntents;
|
||||
|
||||
internal class ConsumeUploadIntentsActivity(
|
||||
IDatabase database,
|
||||
ILogger<ConsumeUploadIntentsActivity> logger)
|
||||
: IActivity<ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.UploadIntents
|
||||
{
|
||||
public const string ExecuteEndpointName = "consume-upload-intents";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<ConsumeUploadIntentsArguments> executeContext)
|
||||
internal class ConsumeUploadIntentsActivity(
|
||||
IDatabase database,
|
||||
ILogger<ConsumeUploadIntentsActivity> logger)
|
||||
: IActivity<ConsumeUploadIntentsArguments, ConsumeUploadIntentsLog>
|
||||
{
|
||||
var intentIds = executeContext.Arguments.IntentIds;
|
||||
public const string ExecuteEndpointName = "consume-upload-intents";
|
||||
|
||||
var intents = await database.UploadIntents
|
||||
.Where(i => intentIds.Contains(i.Id))
|
||||
.ToListAsync(executeContext.CancellationToken);
|
||||
|
||||
foreach (var intent in intents)
|
||||
intent.Status = UploadIntentStatus.Consumed;
|
||||
|
||||
await database.SaveChangesAsync(executeContext.CancellationToken);
|
||||
|
||||
logger.LogInformation("Marked {Count} upload intent(s) as consumed", intents.Count);
|
||||
|
||||
return executeContext.Completed(new ConsumeUploadIntentsLog(intentIds));
|
||||
}
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<ConsumeUploadIntentsLog> compensateContext)
|
||||
{
|
||||
try
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<ConsumeUploadIntentsArguments> executeContext)
|
||||
{
|
||||
var intentIds = compensateContext.Log.IntentIds;
|
||||
var intentIds = executeContext.Arguments.IntentIds;
|
||||
|
||||
var intents = await database.UploadIntents
|
||||
.Where(i => intentIds.Contains(i.Id))
|
||||
.ToListAsync(compensateContext.CancellationToken);
|
||||
.ToListAsync(executeContext.CancellationToken);
|
||||
|
||||
foreach (var intent in intents)
|
||||
intent.Status = UploadIntentStatus.Issued;
|
||||
intent.Status = UploadIntentStatus.Consumed;
|
||||
|
||||
await database.SaveChangesAsync(compensateContext.CancellationToken);
|
||||
await database.SaveChangesAsync(executeContext.CancellationToken);
|
||||
|
||||
logger.LogInformation("Reverted {Count} upload intent(s) back to Issued", intents.Count);
|
||||
logger.LogInformation("Marked {Count} upload intent(s) as consumed", intents.Count);
|
||||
|
||||
return compensateContext.Compensated();
|
||||
return executeContext.Completed(new ConsumeUploadIntentsLog(intentIds));
|
||||
}
|
||||
catch (Exception exception)
|
||||
|
||||
public async Task<CompensationResult> Compensate(CompensateContext<ConsumeUploadIntentsLog> compensateContext)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate upload intent consumption");
|
||||
return compensateContext.Failed(exception);
|
||||
try
|
||||
{
|
||||
var intentIds = compensateContext.Log.IntentIds;
|
||||
|
||||
var intents = await database.UploadIntents
|
||||
.Where(i => intentIds.Contains(i.Id))
|
||||
.ToListAsync(compensateContext.CancellationToken);
|
||||
|
||||
foreach (var intent in intents)
|
||||
intent.Status = UploadIntentStatus.Issued;
|
||||
|
||||
await database.SaveChangesAsync(compensateContext.CancellationToken);
|
||||
|
||||
logger.LogInformation("Reverted {Count} upload intent(s) back to Issued", intents.Count);
|
||||
|
||||
return compensateContext.Compensated();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to compensate upload intent consumption");
|
||||
return compensateContext.Failed(exception);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user