Simplify infrastructure services and move all recurring jobs to Hangfire

This commit is contained in:
2026-09-24 12:18:57 +02:00
parent 0fa033faff
commit 0cba32f776
24 changed files with 361 additions and 413 deletions
@@ -2,11 +2,12 @@ namespace Musify.Application.Contracts;
public interface IStorageService public interface IStorageService
{ {
/// <summary>Opens an object for reading, or returns null when it does not exist.</summary>
public Task<Stream?> GetFileAsync(string bucket, string key, CancellationToken cancellationToken); public Task<Stream?> GetFileAsync(string bucket, string key, CancellationToken cancellationToken);
public Task<ObjectMetaData?> HeadObjectAsync(string bucket, string key, CancellationToken cancellationToken); public Task<ObjectMetaData?> HeadObjectAsync(string bucket, string key, CancellationToken cancellationToken);
public Task<string> GetUploadUrlAsync(string bucket, string key, string contentType, TimeSpan expirationTime, CancellationToken cancellationToken, bool preventOverwrite = true); public Task<string> GetUploadUrlAsync(string bucket, string key, string contentType, TimeSpan expirationTime, bool preventOverwrite = true, CancellationToken cancellationToken = default);
public Task UploadFileAsync(string filePath, string bucket, string key, CancellationToken cancellationToken); public Task UploadFileAsync(string filePath, string bucket, string key, CancellationToken cancellationToken);
@@ -1,4 +1,3 @@
using ErrorOr;
using Mediator; using Mediator;
using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
@@ -10,17 +9,17 @@ using Musify.Domain.ValueObjects;
namespace Musify.Application.Mixes; namespace Musify.Application.Mixes;
public record GenerateMixesForUserCommand(long UserId) public record GenerateMixesForUserCommand(long UserId)
: ICommand<ErrorOr<Success>>; : ICommand;
public class GenerateMixesForUserCommandHandler( public class GenerateMixesForUserCommandHandler(
IDatabase database, IDatabase database,
MixConfiguration mixConfiguration, MixConfiguration mixConfiguration,
ILogger<GenerateMixesForUserCommandHandler> logger) ILogger<GenerateMixesForUserCommandHandler> logger)
: ICommandHandler<GenerateMixesForUserCommand, ErrorOr<Success>> : ICommandHandler<GenerateMixesForUserCommand>
{ {
private sealed record MixDraft(MixKind Kind, IReadOnlyList<Guid> TrackIds); private sealed record MixDraft(MixKind Kind, IReadOnlyList<Guid> TrackIds);
public async ValueTask<ErrorOr<Success>> Handle(GenerateMixesForUserCommand request, CancellationToken cancellationToken) public async ValueTask<Unit> Handle(GenerateMixesForUserCommand request, CancellationToken cancellationToken)
{ {
var userId = request.UserId; var userId = request.UserId;
@@ -33,7 +32,7 @@ public class GenerateMixesForUserCommandHandler(
if (libraryTrackIds.Count == 0) if (libraryTrackIds.Count == 0)
{ {
logger.LogInformation("User {UserId} has no tracks yet, skipping mix generation", userId); logger.LogInformation("User {UserId} has no tracks yet, skipping mix generation", userId);
return Result.Success; return Unit.Value;
} }
var historyTrackIds = await database.ListeningHistories var historyTrackIds = await database.ListeningHistories
@@ -63,7 +62,7 @@ public class GenerateMixesForUserCommandHandler(
if (drafts.Count == 0) if (drafts.Count == 0)
{ {
logger.LogInformation("No candidate songs found for user {UserId}, skipping mix generation", userId); logger.LogInformation("No candidate songs found for user {UserId}, skipping mix generation", userId);
return Result.Success; return Unit.Value;
} }
await ReplaceMixesAsync(userId, drafts, cancellationToken); await ReplaceMixesAsync(userId, drafts, cancellationToken);
@@ -71,7 +70,7 @@ public class GenerateMixesForUserCommandHandler(
await database.SaveChangesAsync(cancellationToken); await database.SaveChangesAsync(cancellationToken);
logger.LogInformation("Generated {Count} mixes for user {UserId}", drafts.Count, userId); logger.LogInformation("Generated {Count} mixes for user {UserId}", drafts.Count, userId);
return Result.Success; return Unit.Value;
} }
private List<Guid> Take(IEnumerable<Guid> trackIds) => private List<Guid> Take(IEnumerable<Guid> trackIds) =>
@@ -69,7 +69,7 @@ public sealed class UploadIntentIssuer(
foreach (var item in planned) foreach (var item in planned)
{ {
var uploadUrl = await storageService.GetUploadUrlAsync( var uploadUrl = await storageService.GetUploadUrlAsync(
storageConfiguration.Bucket, item.Key, item.Request.ContentType, expiresIn, cancellationToken); storageConfiguration.Bucket, item.Key, item.Request.ContentType, expiresIn, cancellationToken: cancellationToken);
var intent = new UploadIntent var intent = new UploadIntent
{ {
@@ -0,0 +1,38 @@
using Mediator;
using Microsoft.Extensions.Logging;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
namespace Musify.Application.UploadIntents;
/// <summary>Deletes temporary upload objects that nobody claimed within the retention window.</summary>
public record DeleteStaleTempUploadsCommand : ICommand<int>;
public class DeleteStaleTempUploadsCommandHandler(
IStorageService storageService,
ApplicationStorageConfiguration storageConfiguration,
UploadIntentConfiguration uploadIntentConfiguration,
ILogger<DeleteStaleTempUploadsCommandHandler> logger)
: ICommandHandler<DeleteStaleTempUploadsCommand, int>
{
public async ValueTask<int> Handle(DeleteStaleTempUploadsCommand request, CancellationToken cancellationToken)
{
var cutOff = DateTime.UtcNow.AddDays(-uploadIntentConfiguration.TempUploadsRetentionDays);
var deleted = 0;
await foreach (var (key, lastModified) in storageService.ListObjectsAsync(
storageConfiguration.Bucket, uploadIntentConfiguration.TempRootPrefix, cancellationToken))
{
if (lastModified >= cutOff)
continue;
await storageService.RemoveFileAsync(storageConfiguration.Bucket, key, cancellationToken);
deleted++;
}
if (deleted > 0)
logger.LogInformation("Deleted {Deleted} orphaned temp uploads", deleted);
return deleted;
}
}
@@ -0,0 +1,46 @@
using Mediator;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
namespace Musify.Application.UploadIntents;
/// <summary>Marks overdue upload intents as expired and deletes the ones expired for longer than the retention window.</summary>
public record ExpireUploadIntentsCommand : ICommand<int>;
public class ExpireUploadIntentsCommandHandler(
IDatabase database,
UploadIntentConfiguration uploadIntentConfiguration,
ILogger<ExpireUploadIntentsCommandHandler> logger)
: ICommandHandler<ExpireUploadIntentsCommand, int>
{
public async ValueTask<int> Handle(ExpireUploadIntentsCommand request, CancellationToken cancellationToken)
{
var now = DateTime.UtcNow;
var expiredIntents = await database.UploadIntents
.Where(intent => intent.Status == UploadIntentStatus.Issued && intent.ExpiresAt < now)
.ToListAsync(cancellationToken);
foreach (var intent in expiredIntents)
intent.Status = UploadIntentStatus.Expired;
var retentionCutoff = now.AddDays(-uploadIntentConfiguration.ExpiredIntentsRetentionDays);
var oldExpiredIntents = await database.UploadIntents
.Where(intent => intent.Status == UploadIntentStatus.Expired && intent.ExpiresAt < retentionCutoff)
.ToListAsync(cancellationToken);
database.UploadIntents.RemoveRange(oldExpiredIntents);
if (expiredIntents.Count == 0 && oldExpiredIntents.Count == 0)
return 0;
await database.SaveChangesAsync(cancellationToken);
logger.LogInformation("Marked {Expired} upload intents as expired and deleted {Deleted} old ones",
expiredIntents.Count, oldExpiredIntents.Count);
return expiredIntents.Count + oldExpiredIntents.Count;
}
}
@@ -23,14 +23,7 @@ public class DailyMixGenerationJob(
logger.LogInformation("Starting daily mix generation for {Count} users", userIds.Count); logger.LogInformation("Starting daily mix generation for {Count} users", userIds.Count);
foreach (var userId in userIds) foreach (var userId in userIds)
{ await mediator.Send(new GenerateMixesForUserCommand(userId), cancellationToken);
var result = await mediator.Send(new GenerateMixesForUserCommand(userId), cancellationToken);
if (result.IsError)
{
logger.LogWarning("Failed to generate mixes for user {UserId}: {Error}",
userId, result.FirstError.Description);
}
}
logger.LogInformation("Finished daily mix generation"); logger.LogInformation("Finished daily mix generation");
} }
@@ -0,0 +1,41 @@
using Hangfire;
using Microsoft.Extensions.Hosting;
using Musify.Application.Configuration;
namespace Musify.Infrastructure.Jobs;
/// <summary>Registers every recurring Hangfire job when the host starts.</summary>
public sealed class RecurringJobsRegistrar(
IRecurringJobManager recurringJobManager,
UploadIntentConfiguration uploadIntentConfiguration)
: IHostedService
{
public Task StartAsync(CancellationToken cancellationToken)
{
var options = new RecurringJobOptions { TimeZone = TimeZoneInfo.Utc };
recurringJobManager.AddOrUpdate<DailyMixGenerationJob>(
"daily-mix-generation", job => job.RunAsync(CancellationToken.None), Cron.Daily(), options);
recurringJobManager.AddOrUpdate<ListeningHistoryCleanupJob>(
"listening-history-cleanup", job => job.RunAsync(CancellationToken.None), Cron.Daily(3), options);
recurringJobManager.AddOrUpdate<UploadIntentExpirationJob>(
"upload-intent-expiration", job => job.RunAsync(CancellationToken.None),
EveryCron(uploadIntentConfiguration.ExpirationJobIntervalSeconds), options);
recurringJobManager.AddOrUpdate<TempUploadsCleanupJob>(
"temp-uploads-cleanup", job => job.RunAsync(CancellationToken.None),
EveryCron(uploadIntentConfiguration.TempCleanupJobIntervalSeconds), options);
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
/// <summary>Hangfire cron has a one-minute resolution, so the interval is rounded up to whole minutes.</summary>
internal static string EveryCron(int intervalSeconds)
{
var minutes = Math.Max(1, (int)Math.Ceiling(intervalSeconds / 60.0));
return minutes < 60
? $"*/{minutes} * * * *"
: $"0 */{Math.Max(1, minutes / 60)} * * *";
}
}
@@ -0,0 +1,10 @@
using Mediator;
using Musify.Application.UploadIntents;
namespace Musify.Infrastructure.Jobs;
public class TempUploadsCleanupJob(IMediator mediator)
{
public async Task RunAsync(CancellationToken cancellationToken) =>
await mediator.Send(new DeleteStaleTempUploadsCommand(), cancellationToken);
}
@@ -1,70 +0,0 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
namespace Musify.Infrastructure.Jobs;
public class TemporalUploadsCleanUpJob(
IServiceScopeFactory scopeFactory,
ILogger<TemporalUploadsCleanUpJob> logger,
ApplicationStorageConfiguration storageConfiguration,
UploadIntentConfiguration uploadIntentConfiguration)
: BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
logger.LogInformation("TemporalUploadsCleanUpJob started (interval: {Interval}s, prefix: {Prefix})",
uploadIntentConfiguration.TempCleanupJobIntervalSeconds,
uploadIntentConfiguration.TempRootPrefix);
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.TempCleanupJobIntervalSeconds));
do
{
try
{
await RunAsync(stoppingToken);
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
logger.LogError(exception, "Error in TemporalUploadsCleanUpJob");
}
}
while (await timer.WaitForNextTickAsync(stoppingToken));
}
private async Task RunAsync(CancellationToken cancellationToken)
{
await using var scope = scopeFactory.CreateAsyncScope();
var storageService = scope.ServiceProvider.GetRequiredService<IStorageService>();
var cutOff = DateTime.UtcNow.AddDays(-uploadIntentConfiguration.TempUploadsRetentionDays);
var deleted = 0;
var errors = 0;
await foreach (var (key, lastModified) in storageService.ListObjectsAsync(
storageConfiguration.Bucket,
uploadIntentConfiguration.TempRootPrefix,
cancellationToken))
{
if (lastModified >= cutOff)
continue;
try
{
await storageService.RemoveFileAsync(storageConfiguration.Bucket, key, cancellationToken);
deleted++;
}
catch (Exception exception)
{
errors++;
logger.LogWarning(exception, "Failed to delete orphaned temp object {Key}", key);
}
}
if (deleted > 0 || errors > 0)
logger.LogInformation("TemporalUploadsCleanUpJob: deleted {Deleted} orphaned objects, {Errors} errors", deleted, errors);
}
}
@@ -1,70 +1,10 @@
using Microsoft.EntityFrameworkCore; using Mediator;
using Microsoft.Extensions.DependencyInjection; using Musify.Application.UploadIntents;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Musify.Application.Configuration;
using Musify.Application.Contracts;
using Musify.Domain.ValueObjects;
namespace Musify.Infrastructure.Jobs; namespace Musify.Infrastructure.Jobs;
public class UploadIntentExpirationJob( public class UploadIntentExpirationJob(IMediator mediator)
IServiceScopeFactory scopeFactory,
ILogger<UploadIntentExpirationJob> logger,
UploadIntentConfiguration uploadIntentConfiguration)
: BackgroundService
{ {
protected override async Task ExecuteAsync(CancellationToken stoppingToken) public async Task RunAsync(CancellationToken cancellationToken) =>
{ await mediator.Send(new ExpireUploadIntentsCommand(), cancellationToken);
logger.LogInformation("UploadIntentExpirationJob started (interval: {Interval}s)", uploadIntentConfiguration.ExpirationJobIntervalSeconds);
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(uploadIntentConfiguration.ExpirationJobIntervalSeconds));
do
{
try
{
await RunAsync(stoppingToken);
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
logger.LogError(exception, "Error in UploadIntentExpirationJob");
}
}
while (await timer.WaitForNextTickAsync(stoppingToken));
}
private async Task RunAsync(CancellationToken cancellationToken)
{
await using var scope = scopeFactory.CreateAsyncScope();
var database = scope.ServiceProvider.GetRequiredService<IDatabase>();
var now = DateTime.UtcNow;
var expiredIntents = await database.UploadIntents
.Where(i => i.Status == UploadIntentStatus.Issued && i.ExpiresAt < now)
.ToListAsync(cancellationToken);
if (expiredIntents.Count > 0)
{
foreach (var intent in expiredIntents)
intent.Status = UploadIntentStatus.Expired;
await database.SaveChangesAsync(cancellationToken);
logger.LogInformation("Marked {Count} upload intents as Expired", expiredIntents.Count);
}
var retentionCutoff = now.AddDays(-uploadIntentConfiguration.ExpiredIntentsRetentionDays);
var oldExpiredIntents = await database.UploadIntents
.Where(i => i.Status == UploadIntentStatus.Expired && i.ExpiresAt < retentionCutoff)
.ToListAsync(cancellationToken);
if (oldExpiredIntents.Count > 0)
{
database.UploadIntents.RemoveRange(oldExpiredIntents);
await database.SaveChangesAsync(cancellationToken);
logger.LogInformation("Deleted {Count} old Expired upload intents past retention window", oldExpiredIntents.Count);
}
}
} }
@@ -6,10 +6,6 @@ namespace Musify.Infrastructure.MassTransit;
public class MassTransitEventBus(IPublishEndpoint publishEndpoint) public class MassTransitEventBus(IPublishEndpoint publishEndpoint)
: IEventBus : IEventBus
{ {
public async Task PublishAsync<T>(T message, CancellationToken cancellationToken) where T : class public Task PublishAsync<T>(T message, CancellationToken cancellationToken) where T : class =>
{ publishEndpoint.Publish(message, cancellationToken);
ArgumentNullException.ThrowIfNull(message, nameof(message));
await publishEndpoint.Publish(message, cancellationToken);
}
} }
@@ -7,53 +7,41 @@ using Musify.Infrastructure.Configuration;
namespace Musify.Infrastructure.Services; namespace Musify.Infrastructure.Services;
public partial class AudioService(ILogger<AudioService> logger, public sealed partial class AudioTranscoderService(
ILogger<AudioTranscoderService> logger,
AudioConfiguration audioConfiguration) AudioConfiguration audioConfiguration)
: IAudioTranscoderService : IAudioTranscoderService
{ {
private record FfmpegExecutionResult(int ExitCode, string StandardOutput, string StandardError); private sealed record FfmpegExecutionResult(int ExitCode, string StandardOutput, string StandardError);
[GeneratedRegex(@"Duration:\s*(\d+):(\d{2}):(\d{2}(?:\.\d+)?)", RegexOptions.IgnoreCase)] [GeneratedRegex(@"Duration:\s*(\d+):(\d{2}):(\d{2}(?:\.\d+)?)", RegexOptions.IgnoreCase)]
private static partial Regex DurationRegex(); private static partial Regex DurationRegex();
public async Task<AudioTranscodeResult> TranscodeToAudioFileAsync(Stream audioStream, string destinationPath, CancellationToken cancellationToken) public async Task<AudioTranscodeResult> TranscodeToAudioFileAsync(Stream audioStream, string destinationPath, CancellationToken cancellationToken)
{ {
try Directory.CreateDirectory(destinationPath);
logger.LogInformation("Starting audio transcoding for {DestinationPath}", destinationPath);
var executionResult = await ExecuteFfmpegAsync(
BuildAudioFileArguments(destinationPath, audioConfiguration.Ffmpeg),
destinationPath,
audioStream,
"TranscodeToAudioFile",
cancellationToken);
if (executionResult.ExitCode != 0)
{ {
Directory.CreateDirectory(destinationPath); logger.LogError("Audio transcoding failed with exit code {ExitCode}. Output: {StandardOutput}, Error: {StandardError}",
executionResult.ExitCode, executionResult.StandardOutput, executionResult.StandardError);
logger.LogInformation("Starting audio transcoding for {DestinationPath}", destinationPath); return new AudioTranscodeResult(executionResult.ExitCode, TimeSpan.Zero);
var executionResult = await ExecuteFfmpegWithInputAsync(
BuildAudioFileArguments(destinationPath, audioConfiguration.Ffmpeg),
destinationPath,
audioStream,
"TranscodeToAudioFile",
cancellationToken);
if (executionResult.ExitCode != 0)
{
logger.LogError("Audio transcoding failed with exit code {ExitCode}. Output: {StandardOutput}, Error: {StandardError}",
executionResult.ExitCode, executionResult.StandardOutput, executionResult.StandardError);
return new AudioTranscodeResult(executionResult.ExitCode, TimeSpan.Zero);
}
logger.LogInformation("Audio transcoding completed for {DestinationPath}", destinationPath);
var outputPath = Path.Combine(destinationPath, audioConfiguration.Ffmpeg.OutputFileName);
var duration = await GetAudioDurationAsync(outputPath, cancellationToken);
return new AudioTranscodeResult(executionResult.ExitCode, duration);
}
catch (TimeoutException timeoutException)
{
logger.LogError(timeoutException, "Audio transcoding timed out after {Timeout} for {DestinationPath}", audioConfiguration.TranscodingTimeout, destinationPath);
throw;
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to transcode audio for {DestinationPath}", destinationPath);
throw;
} }
logger.LogInformation("Audio transcoding completed for {DestinationPath}", destinationPath);
var outputPath = Path.Combine(destinationPath, audioConfiguration.Ffmpeg.OutputFileName);
var duration = await GetAudioDurationAsync(outputPath, cancellationToken);
return new AudioTranscodeResult(executionResult.ExitCode, duration);
} }
private async Task<TimeSpan> GetAudioDurationAsync(string filePath, CancellationToken cancellationToken) private async Task<TimeSpan> GetAudioDurationAsync(string filePath, CancellationToken cancellationToken)
@@ -61,6 +49,7 @@ public partial class AudioService(ILogger<AudioService> logger,
var executionResult = await ExecuteFfmpegAsync( var executionResult = await ExecuteFfmpegAsync(
BuildProbeDurationArguments(filePath), BuildProbeDurationArguments(filePath),
Path.GetDirectoryName(filePath) ?? Environment.CurrentDirectory, Path.GetDirectoryName(filePath) ?? Environment.CurrentDirectory,
input: null,
"ProbeAudioDuration", "ProbeAudioDuration",
cancellationToken); cancellationToken);
@@ -75,37 +64,14 @@ public partial class AudioService(ILogger<AudioService> logger,
return new TimeSpan(hours, minutes, 0) + TimeSpan.FromSeconds(seconds); return new TimeSpan(hours, minutes, 0) + TimeSpan.FromSeconds(seconds);
} }
private async Task<FfmpegExecutionResult> ExecuteFfmpegWithInputAsync(
string arguments,
string workingDirectory,
Stream inputStream,
string operationName,
CancellationToken cancellationToken)
{
using var process = BuildProcess(arguments, workingDirectory, redirectStandardInput: true);
process.Start();
logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory);
var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
var errorTask = process.StandardError.ReadToEndAsync(cancellationToken);
if (inputStream.CanSeek)
inputStream.Position = 0;
await inputStream.CopyToAsync(process.StandardInput.BaseStream, cancellationToken);
process.StandardInput.Close();
return await WaitForProcessResultAsync(process, outputTask, errorTask, operationName, cancellationToken);
}
private async Task<FfmpegExecutionResult> ExecuteFfmpegAsync( private async Task<FfmpegExecutionResult> ExecuteFfmpegAsync(
string arguments, string arguments,
string workingDirectory, string workingDirectory,
Stream? input,
string operationName, string operationName,
CancellationToken cancellationToken) CancellationToken cancellationToken)
{ {
using var process = BuildProcess(arguments, workingDirectory, redirectStandardInput: false); using var process = BuildProcess(arguments, workingDirectory, redirectStandardInput: input != null);
process.Start(); process.Start();
logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory); logger.LogDebug("[{Operation}] ffmpeg started (PID: {ProcessId}) in {WorkingDirectory}", operationName, process.Id, workingDirectory);
@@ -113,6 +79,15 @@ public partial class AudioService(ILogger<AudioService> logger,
var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken); var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
var errorTask = process.StandardError.ReadToEndAsync(cancellationToken); var errorTask = process.StandardError.ReadToEndAsync(cancellationToken);
if (input != null)
{
if (input.CanSeek)
input.Position = 0;
await input.CopyToAsync(process.StandardInput.BaseStream, cancellationToken);
process.StandardInput.Close();
}
return await WaitForProcessResultAsync(process, outputTask, errorTask, operationName, cancellationToken); return await WaitForProcessResultAsync(process, outputTask, errorTask, operationName, cancellationToken);
} }
@@ -148,11 +123,8 @@ public partial class AudioService(ILogger<AudioService> logger,
} }
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{ {
logger.LogError("[{Operation}] ffmpeg timed out after {Timeout} (PID: {ProcessId})",
operationName, audioConfiguration.TranscodingTimeout, process.Id);
TryKillProcess(process); TryKillProcess(process);
throw new TimeoutException($"ffmpeg execution exceeded timeout {audioConfiguration.TranscodingTimeout}."); throw new TimeoutException($"ffmpeg execution exceeded timeout {audioConfiguration.TranscodingTimeout} (PID: {process.Id}).");
} }
var standardError = await errorTask; var standardError = await errorTask;
@@ -171,9 +143,9 @@ public partial class AudioService(ILogger<AudioService> logger,
if (!process.HasExited) if (!process.HasExited)
process.Kill(); process.Kill();
} }
catch (Exception exception) catch (InvalidOperationException exception)
{ {
logger.LogError(exception, "Failed to kill ffmpeg process (PID: {ProcessId})", process.Id); logger.LogWarning(exception, "Could not kill ffmpeg process (PID: {ProcessId}); it had already exited", process.Id);
} }
} }
@@ -10,30 +10,22 @@ public class PictureService(ILogger<PictureService> logger)
{ {
public async Task<Stream> ResizePictureAsWebpAsync(Stream pictureStream, int width, int height, CancellationToken cancellationToken) public async Task<Stream> ResizePictureAsWebpAsync(Stream pictureStream, int width, int height, CancellationToken cancellationToken)
{ {
try logger.LogDebug("Resizing picture to {Width}x{Height}", width, height);
if (pictureStream.CanSeek && pictureStream.Position > 0)
pictureStream.Position = 0;
using var picture = await Image.LoadAsync(pictureStream, cancellationToken);
picture.Mutate(options => options.Resize(new ResizeOptions
{ {
logger.LogDebug("Resizing picture to {Width}x{Height}", width, height); Size = new Size(width, height),
if (pictureStream.CanSeek && pictureStream.Position > 0) Mode = ResizeMode.Max
pictureStream.Position = 0; }));
using var picture = await Image.LoadAsync(pictureStream, cancellationToken); var memoryStream = new MemoryStream();
picture.Mutate(options => options.Resize(new ResizeOptions await picture.SaveAsWebpAsync(memoryStream, cancellationToken);
{ memoryStream.Position = 0;
Size = new Size(width, height),
Mode = ResizeMode.Max
}));
var memoryStream = new MemoryStream(); logger.LogDebug("Picture resized to {Width}x{Height}, length: {Length}", width, height, memoryStream.Length);
await picture.SaveAsWebpAsync(memoryStream, cancellationToken); return memoryStream;
memoryStream.Position = 0;
logger.LogDebug("Picture resized to {Width}x{Height}, length: {Length}", width, height, memoryStream.Length);
return memoryStream;
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to resize picture to {Width}x{Height}", width, height);
throw;
}
} }
} }
@@ -28,7 +28,7 @@ public static class ServicesDependencyInjection
return new AmazonS3Client(configuration.AccessKey, configuration.SecretAccessKey, s3Configuration); return new AmazonS3Client(configuration.AccessKey, configuration.SecretAccessKey, s3Configuration);
}); });
services.AddScoped<IStorageService, StorageService>(); services.AddSingleton<IStorageService, StorageService>();
return services; return services;
} }
@@ -45,17 +45,7 @@ public static class ServicesDependencyInjection
return services; return services;
} }
public static IServiceCollection AddUploadIntentJobs(this IServiceCollection services, IConfiguration configuration) public static IServiceCollection AddJobs(this IServiceCollection services)
{
services.AddUploadIntentConfiguration(configuration);
services.AddHostedService<UploadIntentExpirationJob>();
services.AddHostedService<TemporalUploadsCleanUpJob>();
return services;
}
public static IServiceCollection AddDailyMixGenerationJob(this IServiceCollection services, IConfiguration configuration)
{ {
services.AddHangfire((serviceProvider, hangfireConfiguration) => services.AddHangfire((serviceProvider, hangfireConfiguration) =>
{ {
@@ -67,6 +57,9 @@ public static class ServicesDependencyInjection
services.AddHangfireServer(); services.AddHangfireServer();
services.AddScoped<DailyMixGenerationJob>(); services.AddScoped<DailyMixGenerationJob>();
services.AddScoped<ListeningHistoryCleanupJob>(); services.AddScoped<ListeningHistoryCleanupJob>();
services.AddScoped<UploadIntentExpirationJob>();
services.AddScoped<TempUploadsCleanupJob>();
services.AddHostedService<RecurringJobsRegistrar>();
return services; return services;
} }
@@ -83,7 +76,7 @@ public static class ServicesDependencyInjection
{ {
services.AddValidatedOptions<AudioConfiguration>(configuration); services.AddValidatedOptions<AudioConfiguration>(configuration);
services.AddScoped<IAudioTranscoderService, AudioService>(); services.AddScoped<IAudioTranscoderService, AudioTranscoderService>();
return services; return services;
} }
} }
@@ -1,29 +1,27 @@
using System.Runtime.CompilerServices;
using Amazon.S3; using Amazon.S3;
using Amazon.S3.Model; using Amazon.S3.Model;
using Amazon.S3.Transfer; using Amazon.S3.Transfer;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using MimeMapping; using MimeMapping;
using Musify.Application.Contracts; using Musify.Application.Contracts;
using Musify.Application.Shared;
using Musify.Infrastructure.Configuration; using Musify.Infrastructure.Configuration;
using ObjectMetaData = Musify.Application.Contracts.ObjectMetaData;
namespace Musify.Infrastructure.Services; namespace Musify.Infrastructure.Services;
public class StorageService(IAmazonS3 amazonS3, ILogger<StorageService> logger, public sealed class StorageService(
IAmazonS3 amazonS3,
ILogger<StorageService> logger,
InfrastructureStorageConfiguration storageClientConfiguration) InfrastructureStorageConfiguration storageClientConfiguration)
: IStorageService : IStorageService
{ {
public async Task<Stream?> GetFileAsync(string bucket, string key, CancellationToken cancellationToken) public async Task<Stream?> GetFileAsync(string bucket, string key, CancellationToken cancellationToken)
{ {
var request = new GetObjectRequest
{
BucketName = bucket,
Key = key,
};
try try
{ {
var response = await amazonS3.GetObjectAsync(request, cancellationToken); var response = await amazonS3.GetObjectAsync(
new GetObjectRequest { BucketName = bucket, Key = key }, cancellationToken);
logger.LogDebug("Retrieved file from S3 {Bucket}/{Key}", bucket, key); logger.LogDebug("Retrieved file from S3 {Bucket}/{Key}", bucket, key);
return response.ResponseStream; return response.ResponseStream;
} }
@@ -32,11 +30,6 @@ public class StorageService(IAmazonS3 amazonS3, ILogger<StorageService> logger,
logger.LogDebug("File not found in S3 {Bucket}/{Key}", bucket, key); logger.LogDebug("File not found in S3 {Bucket}/{Key}", bucket, key);
return null; return null;
} }
catch (Exception exception)
{
logger.LogError(exception, "Failed to get file from S3 {Bucket}/{Key}", bucket, key);
throw;
}
} }
public async Task<string> GetUploadUrlAsync( public async Task<string> GetUploadUrlAsync(
@@ -44,8 +37,8 @@ public class StorageService(IAmazonS3 amazonS3, ILogger<StorageService> logger,
string key, string key,
string contentType, string contentType,
TimeSpan expirationTime, TimeSpan expirationTime,
CancellationToken cancellationToken, bool preventOverwrite = true,
bool preventOverwrite = true) CancellationToken cancellationToken = default)
{ {
var request = new GetPreSignedUrlRequest var request = new GetPreSignedUrlRequest
{ {
@@ -71,140 +64,81 @@ public class StorageService(IAmazonS3 amazonS3, ILogger<StorageService> logger,
var response = await amazonS3.GetObjectMetadataAsync(bucket, key, cancellationToken); var response = await amazonS3.GetObjectMetadataAsync(bucket, key, cancellationToken);
return new ObjectMetaData(response.Headers.ContentType, response.ContentLength); return new ObjectMetaData(response.Headers.ContentType, response.ContentLength);
} }
catch (AmazonS3Exception ex) when (ex.StatusCode == System.Net.HttpStatusCode.NotFound) catch (AmazonS3Exception exception) when (exception.StatusCode == System.Net.HttpStatusCode.NotFound)
{ {
return null; return null;
} }
catch (Exception exception)
{
logger.LogError(exception, "HEAD failed for {Bucket}/{Key}", bucket, key);
throw;
}
} }
public async Task RemoveFileAsync(string bucket, string key, CancellationToken cancellationToken) public async Task RemoveFileAsync(string bucket, string key, CancellationToken cancellationToken)
{ {
var request = new DeleteObjectRequest
{
BucketName = bucket,
Key = key,
};
logger.LogDebug("Removing file from S3 {Bucket}/{Key}", bucket, key); logger.LogDebug("Removing file from S3 {Bucket}/{Key}", bucket, key);
await amazonS3.DeleteObjectAsync(request, cancellationToken); await amazonS3.DeleteObjectAsync(new DeleteObjectRequest { BucketName = bucket, Key = key }, cancellationToken);
} }
public async Task<IReadOnlyList<string>> TransferFilesAsync(string sourceDirectory, string bucket, string route, CancellationToken cancellationToken) public async Task<IReadOnlyList<string>> TransferFilesAsync(
string sourceDirectory, string bucket, string route, CancellationToken cancellationToken)
{ {
var trasnsferUtility = new TransferUtility(amazonS3); using var transferUtility = new TransferUtility(amazonS3);
var files = Directory.GetFiles(sourceDirectory, "*");
var uploadedKeys = new List<string>(); var uploadedKeys = new List<string>();
var successCount = 0;
var failedCount = 0;
foreach (var file in files) foreach (var filePath in Directory.GetFiles(sourceDirectory, "*"))
{ {
string fileName = Path.GetFileName(file); var key = StorageKey.Combine(route, Path.GetFileName(filePath));
string filePath = Path.Combine(sourceDirectory, file);
var key = string.Join('/', new[] { route, fileName } await transferUtility.UploadAsync(
.Where(static s => !string.IsNullOrWhiteSpace(s)) new TransferUtilityUploadRequest
.Select(static s => s.Trim().Trim('/', '\\'))); {
var contentType = MimeUtility.GetMimeMapping(filePath); FilePath = filePath,
try BucketName = bucket,
{ Key = key,
await trasnsferUtility.UploadAsync( ContentType = MimeUtility.GetMimeMapping(filePath)
new TransferUtilityUploadRequest },
{ cancellationToken);
FilePath = filePath,
BucketName = bucket, uploadedKeys.Add(key);
Key = key, logger.LogDebug("Transferred file to S3 {Bucket}/{Key}", bucket, key);
ContentType = contentType
},
cancellationToken);
uploadedKeys.Add(key);
successCount++;
logger.LogDebug("Transferred file to S3 {Bucket}/{Key}", bucket, key);
}
catch (Exception exception)
{
failedCount++;
logger.LogError(exception, "Failed to transfer file to S3 {Bucket}/{Key}", bucket, key);
throw;
}
} }
logger.LogInformation("File transfer completed to {Bucket}/{Route}. Success: {SuccessCount}, Failed: {FailedCount}", bucket, route, successCount, failedCount); logger.LogInformation("Transferred {Count} files to {Bucket}/{Route}", uploadedKeys.Count, bucket, route);
return uploadedKeys; return uploadedKeys;
} }
public async Task UploadFileAsync(string filePath, string bucket, string key, CancellationToken cancellationToken) public async Task UploadFileAsync(string filePath, string bucket, string key, CancellationToken cancellationToken)
{ {
try await using var fileStream = File.OpenRead(filePath);
{ await UploadFileAsync(fileStream, MimeUtility.GetMimeMapping(filePath), bucket, key, cancellationToken);
using var fileStream = File.OpenRead(filePath);
var contentType = MimeUtility.GetMimeMapping(filePath);
await UploadFileAsync(fileStream, contentType, bucket, key, cancellationToken);
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to upload file to S3 {Bucket}/{Key}", bucket, key);
throw;
}
} }
public async Task UploadFileAsync(Stream sourceStream, string contentType, string bucket, string key, CancellationToken cancellationToken) public async Task UploadFileAsync(Stream sourceStream, string contentType, string bucket, string key, CancellationToken cancellationToken)
{ {
var request = new PutObjectRequest await amazonS3.PutObjectAsync(
{ new PutObjectRequest { BucketName = bucket, ContentType = contentType, InputStream = sourceStream, Key = key },
BucketName = bucket, cancellationToken);
ContentType = contentType, logger.LogDebug("Uploaded file to S3 {Bucket}/{Key}", bucket, key);
InputStream = sourceStream,
Key = key
};
try
{
await amazonS3.PutObjectAsync(request, cancellationToken);
logger.LogDebug("Uploaded file to S3 {Bucket}/{Key}", bucket, key);
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to upload file to S3 {Bucket}/{Key}", bucket, key);
throw;
}
} }
public async Task CopyFileAsync(string sourceBucket, string sourceKey, string destinationBucket, string destinationKey, public async Task CopyFileAsync(
CancellationToken cancellationToken) string sourceBucket, string sourceKey, string destinationBucket, string destinationKey, CancellationToken cancellationToken)
{ {
var request = new CopyObjectRequest await amazonS3.CopyObjectAsync(
{ new CopyObjectRequest
SourceBucket = sourceBucket, {
SourceKey = sourceKey, SourceBucket = sourceBucket,
DestinationBucket = destinationBucket, SourceKey = sourceKey,
DestinationKey = destinationKey DestinationBucket = destinationBucket,
}; DestinationKey = destinationKey
},
cancellationToken);
try logger.LogDebug("Copied file from {SourceBucket}/{SourceKey} to {DestinationBucket}/{DestinationKey}",
{ sourceBucket, sourceKey, destinationBucket, destinationKey);
await amazonS3.CopyObjectAsync(request, cancellationToken);
logger.LogInformation("Copied file from {SourceBucket}/{SourceKey} to {DestinationBucket}/{DestinationKey}",
sourceBucket, sourceKey, destinationBucket, destinationKey);
}
catch (Exception exception)
{
logger.LogError(exception, "Failed to copy file from {SourceBucket}/{SourceKey} to {DestinationBucket}/{DestinationKey}",
sourceBucket, sourceKey, destinationBucket, destinationKey);
throw;
}
} }
public async IAsyncEnumerable<StorageObject> ListObjectsAsync( public async IAsyncEnumerable<StorageObject> ListObjectsAsync(
string bucket, string prefix, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) string bucket, string prefix, [EnumeratorCancellation] CancellationToken cancellationToken)
{ {
var listRequest = new ListObjectsV2Request var listRequest = new ListObjectsV2Request
{ {
@@ -221,6 +155,6 @@ public class StorageService(IAmazonS3 amazonS3, ILogger<StorageService> logger,
listRequest.ContinuationToken = listResponse.NextContinuationToken; listRequest.ContinuationToken = listResponse.NextContinuationToken;
} }
while (listResponse.IsTruncated.HasValue && listResponse.IsTruncated.Value); while (listResponse.IsTruncated == true);
} }
} }
@@ -4,6 +4,7 @@ using System.Text.Json.Serialization;
using Mediator; using Mediator;
using Microsoft.AspNetCore.Authentication.JwtBearer; using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.Extensions.Caching.Memory; using Microsoft.Extensions.Caching.Memory;
using Microsoft.IdentityModel.Protocols.OpenIdConnect;
using Musify.Application.Users; using Musify.Application.Users;
namespace Musify.Api.Authentication; namespace Musify.Api.Authentication;
@@ -87,15 +88,15 @@ public sealed class JwtBearerEventsHandler(
try try
{ {
var userInfoEndpoint = await ResolveUserInfoEndpointAsync(tokenValidatedContext, cancellationToken);
var client = httpClientFactory.CreateClient(); var client = httpClientFactory.CreateClient();
var request = new HttpRequestMessage( using var request = new HttpRequestMessage(HttpMethod.Get, userInfoEndpoint)
HttpMethod.Get,
$"{authenticationConfiguration.IssuerAddress.TrimEnd('/')}/oidc/v1/userinfo")
{ {
Headers = { Authorization = new AuthenticationHeaderValue("Bearer", accessToken) } Headers = { Authorization = new AuthenticationHeaderValue("Bearer", accessToken) }
}; };
var response = await client.SendAsync(request, cancellationToken); using var response = await client.SendAsync(request, cancellationToken);
if (!response.IsSuccessStatusCode) if (!response.IsSuccessStatusCode)
{ {
logger.LogDebug("Userinfo request returned {StatusCode}", response.StatusCode); logger.LogDebug("Userinfo request returned {StatusCode}", response.StatusCode);
@@ -111,6 +112,20 @@ public sealed class JwtBearerEventsHandler(
} }
} }
private async Task<string> ResolveUserInfoEndpointAsync(
TokenValidatedContext tokenValidatedContext, CancellationToken cancellationToken)
{
var configurationManager = tokenValidatedContext.Options.ConfigurationManager;
if (configurationManager != null)
{
var configuration = await configurationManager.GetConfigurationAsync(cancellationToken);
if (configuration is OpenIdConnectConfiguration { UserInfoEndpoint: { Length: > 0 } endpoint })
return endpoint;
}
return $"{authenticationConfiguration.IssuerAddress.TrimEnd('/')}/oidc/v1/userinfo";
}
public override Task AuthenticationFailed(AuthenticationFailedContext context) public override Task AuthenticationFailed(AuthenticationFailedContext context)
{ {
logger.LogWarning(context.Exception, "JWT authentication failed"); logger.LogWarning(context.Exception, "JWT authentication failed");
+2 -19
View File
@@ -1,6 +1,4 @@
using Hangfire;
using Musify.Application.Configuration; using Musify.Application.Configuration;
using Musify.Infrastructure.Jobs;
using Musify.Infrastructure.MassTransit; using Musify.Infrastructure.MassTransit;
using Musify.Infrastructure.Persistence; using Musify.Infrastructure.Persistence;
using Musify.Infrastructure.Services; using Musify.Infrastructure.Services;
@@ -26,25 +24,10 @@ services.AddAudioTranscoder(configuration);
services.AddPictureService(); services.AddPictureService();
services.AddDatabase(configuration); services.AddDatabase(configuration);
services.AddMassTransitConsumers(configuration); services.AddMassTransitConsumers(configuration);
services.AddUploadIntentJobs(configuration); services.AddUploadIntentConfiguration(configuration);
services.AddMediator(); services.AddMediator();
services.AddDailyMixGenerationJob(configuration); services.AddJobs();
var host = builder.Build(); var host = builder.Build();
using (var scope = host.Services.CreateScope())
{
var recurringJobManager = scope.ServiceProvider.GetRequiredService<IRecurringJobManager>();
recurringJobManager.AddOrUpdate<DailyMixGenerationJob>(
"daily-mix-generation",
job => job.RunAsync(CancellationToken.None),
Cron.Daily(),
new RecurringJobOptions { TimeZone = TimeZoneInfo.Utc });
recurringJobManager.AddOrUpdate<ListeningHistoryCleanupJob>(
"listening-history-cleanup",
job => job.RunAsync(CancellationToken.None),
Cron.Daily(3),
new RecurringJobOptions { TimeZone = TimeZoneInfo.Utc });
}
host.Run(); host.Run();
@@ -18,7 +18,7 @@ public sealed class AlbumEndpointsTests(ApiTestFixture fixture)
private async Task<Guid> CreatePictureIntentAsync(HttpClient client) private async Task<Guid> CreatePictureIntentAsync(HttpClient client)
{ {
fixture.StorageService fixture.StorageService
.GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>()) .GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<bool>(), Arg.Any<CancellationToken>())
.Returns("https://storage.musify.test/presigned-upload"); .Returns("https://storage.musify.test/presigned-upload");
fixture.StorageService fixture.StorageService
.HeadObjectAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<CancellationToken>()) .HeadObjectAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<CancellationToken>())
@@ -17,7 +17,7 @@ public sealed class RequestAlbumPictureUploadCommandHandlerTests : HandlerTestBa
private RequestAlbumPictureUploadCommandHandler CreateHandler(UploadIntentConfiguration? uploadIntentConfig = null) private RequestAlbumPictureUploadCommandHandler CreateHandler(UploadIntentConfiguration? uploadIntentConfig = null)
{ {
storageService storageService
.GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>()) .GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<bool>(), Arg.Any<CancellationToken>())
.Returns("https://storage.musify.test/presigned-upload"); .Returns("https://storage.musify.test/presigned-upload");
var config = uploadIntentConfig ?? TestConfigurations.UploadIntent(); var config = uploadIntentConfig ?? TestConfigurations.UploadIntent();
@@ -17,9 +17,7 @@ public sealed class GenerateMixesForUserCommandHandlerTests : HandlerTestBase
var user = TestEntities.User(); var user = TestEntities.User();
await SeedAsync(user); await SeedAsync(user);
var result = await CreateHandler().Handle(new GenerateMixesForUserCommand(user.Id), TestContext.Current.CancellationToken); await CreateHandler().Handle(new GenerateMixesForUserCommand(user.Id), TestContext.Current.CancellationToken);
Assert.False(result.IsError);
Assert.Empty(await Database.Mixes.ToListAsync(TestContext.Current.CancellationToken)); Assert.Empty(await Database.Mixes.ToListAsync(TestContext.Current.CancellationToken));
} }
@@ -32,9 +30,7 @@ public sealed class GenerateMixesForUserCommandHandlerTests : HandlerTestBase
user, track, user, track,
TestEntities.ListeningHistory(user.Id, track.Id)); TestEntities.ListeningHistory(user.Id, track.Id));
var result = await CreateHandler().Handle(new GenerateMixesForUserCommand(user.Id), TestContext.Current.CancellationToken); await CreateHandler().Handle(new GenerateMixesForUserCommand(user.Id), TestContext.Current.CancellationToken);
Assert.False(result.IsError);
var mixes = await Database.Mixes.ToListAsync(TestContext.Current.CancellationToken); var mixes = await Database.Mixes.ToListAsync(TestContext.Current.CancellationToken);
Assert.Single(mixes); Assert.Single(mixes);
Assert.Equal(MixKind.Daily, mixes[0].Kind); Assert.Equal(MixKind.Daily, mixes[0].Kind);
@@ -54,9 +50,7 @@ public sealed class GenerateMixesForUserCommandHandlerTests : HandlerTestBase
user, listenedTrack, unheardTrack, staleMix, staleItem, user, listenedTrack, unheardTrack, staleMix, staleItem,
TestEntities.ListeningHistory(user.Id, listenedTrack.Id)); TestEntities.ListeningHistory(user.Id, listenedTrack.Id));
var result = await CreateHandler().Handle(new GenerateMixesForUserCommand(user.Id), TestContext.Current.CancellationToken); await CreateHandler().Handle(new GenerateMixesForUserCommand(user.Id), TestContext.Current.CancellationToken);
Assert.False(result.IsError);
var mixes = await Database.Mixes.ToListAsync(TestContext.Current.CancellationToken); var mixes = await Database.Mixes.ToListAsync(TestContext.Current.CancellationToken);
Assert.DoesNotContain(mixes, mix => mix.Id == staleMix.Id); Assert.DoesNotContain(mixes, mix => mix.Id == staleMix.Id);
Assert.Contains(mixes, mix => mix.Kind == MixKind.Discovery); Assert.Contains(mixes, mix => mix.Kind == MixKind.Discovery);
@@ -17,7 +17,7 @@ public sealed class RequestPlayListPictureUploadCommandHandlerTests : HandlerTes
private RequestPlayListPictureUploadCommandHandler CreateHandler(UploadIntentConfiguration? uploadIntentConfig = null) private RequestPlayListPictureUploadCommandHandler CreateHandler(UploadIntentConfiguration? uploadIntentConfig = null)
{ {
storageService storageService
.GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>()) .GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<bool>(), Arg.Any<CancellationToken>())
.Returns("https://storage.musify.test/presigned-upload"); .Returns("https://storage.musify.test/presigned-upload");
var config = uploadIntentConfig ?? TestConfigurations.UploadIntent(); var config = uploadIntentConfig ?? TestConfigurations.UploadIntent();
@@ -18,7 +18,7 @@ public sealed class RequestTrackUploadUrlsCommandHandlerTests : HandlerTestBase
private RequestTrackUploadUrlsCommandHandler CreateHandler(UploadIntentConfiguration? uploadIntentConfig = null) private RequestTrackUploadUrlsCommandHandler CreateHandler(UploadIntentConfiguration? uploadIntentConfig = null)
{ {
storageService storageService
.GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>()) .GetUploadUrlAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<TimeSpan>(), Arg.Any<bool>(), Arg.Any<CancellationToken>())
.Returns("https://storage.musify.test/presigned-upload"); .Returns("https://storage.musify.test/presigned-upload");
var config = uploadIntentConfig ?? TestConfigurations.UploadIntent(); var config = uploadIntentConfig ?? TestConfigurations.UploadIntent();
@@ -0,0 +1,71 @@
using Microsoft.EntityFrameworkCore;
using Musify.Application.Contracts;
using Musify.Application.Tests.TestSupport;
using Musify.Application.UploadIntents;
using Musify.Domain.ValueObjects;
using NSubstitute;
using Xunit;
namespace Musify.Application.Tests.UploadIntents;
public sealed class UploadIntentJobsCommandHandlerTests : HandlerTestBase
{
[Fact]
public async Task ExpireUploadIntents_OverdueIssuedIntent_IsMarkedExpired()
{
var owner = TestEntities.User();
var overdue = TestEntities.UploadIntent(owner.Id, expiresAt: DateTime.UtcNow.AddMinutes(-5));
var active = TestEntities.UploadIntent(owner.Id, expiresAt: DateTime.UtcNow.AddMinutes(5));
await SeedAsync(owner, overdue, active);
var handler = new ExpireUploadIntentsCommandHandler(
Database, TestConfigurations.UploadIntent(), NoOpLogger<ExpireUploadIntentsCommandHandler>());
var changed = await handler.Handle(new ExpireUploadIntentsCommand(), TestContext.Current.CancellationToken);
Assert.Equal(1, changed);
var stored = await Database.UploadIntents.AsNoTracking().ToListAsync(TestContext.Current.CancellationToken);
Assert.Equal(UploadIntentStatus.Expired, stored.Single(intent => intent.Id == overdue.Id).Status);
Assert.Equal(UploadIntentStatus.Issued, stored.Single(intent => intent.Id == active.Id).Status);
}
[Fact]
public async Task ExpireUploadIntents_IntentExpiredLongAgo_IsDeleted()
{
var owner = TestEntities.User();
var old = TestEntities.UploadIntent(
owner.Id, status: UploadIntentStatus.Expired, expiresAt: DateTime.UtcNow.AddDays(-30));
await SeedAsync(owner, old);
var handler = new ExpireUploadIntentsCommandHandler(
Database, TestConfigurations.UploadIntent(), NoOpLogger<ExpireUploadIntentsCommandHandler>());
await handler.Handle(new ExpireUploadIntentsCommand(), TestContext.Current.CancellationToken);
Assert.Empty(await Database.UploadIntents.ToListAsync(TestContext.Current.CancellationToken));
}
[Fact]
public async Task DeleteStaleTempUploads_RemovesOnlyObjectsOlderThanTheRetention()
{
var storage = Substitute.For<IStorageService>();
storage.ListObjectsAsync("test-bucket", "temp", Arg.Any<CancellationToken>()).Returns(Objects(
new StorageObject("temp/old", DateTime.UtcNow.AddDays(-10)),
new StorageObject("temp/new", DateTime.UtcNow)));
var handler = new DeleteStaleTempUploadsCommandHandler(
storage, TestConfigurations.Storage("test-bucket"), TestConfigurations.UploadIntent(),
NoOpLogger<DeleteStaleTempUploadsCommandHandler>());
var deleted = await handler.Handle(new DeleteStaleTempUploadsCommand(), TestContext.Current.CancellationToken);
Assert.Equal(1, deleted);
await storage.Received(1).RemoveFileAsync("test-bucket", "temp/old", Arg.Any<CancellationToken>());
await storage.DidNotReceive().RemoveFileAsync("test-bucket", "temp/new", Arg.Any<CancellationToken>());
}
private static async IAsyncEnumerable<StorageObject> Objects(params StorageObject[] objects)
{
foreach (var item in objects)
yield return item;
await Task.CompletedTask;
}
}
@@ -117,7 +117,7 @@ public sealed class StorageServiceTests(InfrastructureTestFixture fixture) : IAs
var key = $"tests/{Guid.NewGuid():N}/presigned.txt"; var key = $"tests/{Guid.NewGuid():N}/presigned.txt";
var uploadUrl = await service.GetUploadUrlAsync( var uploadUrl = await service.GetUploadUrlAsync(
InfrastructureTestFixture.S3Bucket, key, "text/plain", TimeSpan.FromMinutes(5), TestContext.Current.CancellationToken); InfrastructureTestFixture.S3Bucket, key, "text/plain", TimeSpan.FromMinutes(5), cancellationToken: TestContext.Current.CancellationToken);
using var httpClient = new HttpClient(); using var httpClient = new HttpClient();
using var request = new HttpRequestMessage(HttpMethod.Put, uploadUrl) using var request = new HttpRequestMessage(HttpMethod.Put, uploadUrl)