botris.social/femto-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

media upload

Browse source
This commit is contained in:
john 2025-05-04 21:46:24 +02:00
parent befaa207d7
commit 0d7da2ea85
33 changed files with 257 additions and 353 deletions
Download patch file Download diff file Expand all files Collapse all files

3
Femto.Modules.Media/Contracts/LoadFile/Dto/LoadFileQueryResult.cs Normal file
View file

@ -0,0 +1,3 @@
namespace Femto.Modules.Media.Contracts.LoadFile.Dto;
public record LoadFileQueryResult(Stream Data, string Type, long? Size);

6
Femto.Modules.Media/Contracts/LoadFile/LoadFileQuery.cs Normal file
View file

@ -0,0 +1,6 @@
using Femto.Modules.Media.Contracts.LoadFile.Dto;
using MediatR;
namespace Femto.Modules.Media.Contracts.LoadFile;
public record LoadFileQuery(Guid FileId) : IRequest<LoadFileQueryResult>;

25
Femto.Modules.Media/Contracts/LoadFile/LoadFileQueryHandler.cs Normal file
View file

@ -0,0 +1,25 @@
using Femto.Modules.Media.Contracts.LoadFile.Dto;
using Femto.Modules.Media.Data;
using Femto.Modules.Media.Infrastructure;
using MediatR;
using Microsoft.EntityFrameworkCore;
namespace Femto.Modules.Media.Contracts.LoadFile;
internal class LoadFileQueryHandler(IStorageProvider storage, MediaContext context)
: IRequestHandler<LoadFileQuery, LoadFileQueryResult>
{
public async Task<LoadFileQueryResult> Handle(
LoadFileQuery query,
CancellationToken cancellationToken
)
{
var blob = await context
.SavedBlobs.Where(b => b.Id == query.FileId)
.SingleAsync(cancellationToken: cancellationToken);
var data = await storage.LoadBlob(query.FileId.ToString());
return new(data, blob.Type, blob.Size);
}
}

5
Femto.Modules.Media/Contracts/SaveFile/SaveFileCommand.cs Normal file
View file

@ -0,0 +1,5 @@
using MediatR;
namespace Femto.Modules.Media.Contracts.SaveFile;
public record SaveFileCommand(Stream Data, string ContentType, long? ContentLength) : IRequest<Guid>;

18
Femto.Modules.Media/Contracts/SaveFile/SaveFileCommandHandler.cs Normal file
View file

@ -0,0 +1,18 @@
using Femto.Modules.Media.Data;
using Femto.Modules.Media.Infrastructure;
using MediatR;
namespace Femto.Modules.Media.Contracts.SaveFile;
internal class SaveFileCommandHandler(IStorageProvider storage, MediaContext context)
: IRequestHandler<SaveFileCommand, Guid>
{
public async Task<Guid> Handle(SaveFileCommand command, CancellationToken cancellationToken)
{
var id = Guid.CreateVersion7();
await storage.SaveBlob(id.ToString(), command.Data);
await context.AddAsync(new SavedBlob(id, command.ContentType, command.ContentLength), cancellationToken);
await context.SaveChangesAsync(cancellationToken);
return id;
}
}

5
Femto.Modules.Media/Data/MediaContext.cs
View file

@ -1,16 +1,15 @@
using Femto.Modules.Media.Infrastructure.Integration;
using Microsoft.EntityFrameworkCore;
namespace Femto.Modules.Media.Data;
internal class MediaContext(DbContextOptions<MediaContext> options) : DbContext(options)
{
public virtual DbSet<OutboxEntry> Outbox { get; set; }
public virtual DbSet<SavedBlob> SavedBlobs { get; set; }
protected override void OnModelCreating(ModelBuilder builder)
{
base.OnModelCreating(builder);
builder.HasDefaultSchema("blog");
builder.HasDefaultSchema("media");
builder.ApplyConfigurationsFromAssembly(typeof(MediaContext).Assembly);
}
}

22
Femto.Modules.Media/Data/SavedBlob.cs Normal file
View file

@ -0,0 +1,22 @@
using System.ComponentModel.DataAnnotations.Schema;
namespace Femto.Modules.Media.Data;
[Table("saved_blob")]
internal class SavedBlob
{
public Guid Id { get; private set; }
public DateTime CreatedOn { get; private set; }
public string Type { get; private set; }
public long? Size { get; private set; }
public SavedBlob(Guid id, string type, long? size)
{
Id = id;
CreatedOn = DateTime.UtcNow;
Type = type;
Size = size;
}
private SavedBlob() { }
}

4
Femto.Modules.Media/Femto.Modules.Media.csproj
View file

@ -6,10 +6,6 @@
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<Folder Include="Data\Configurations\" />
</ItemGroup>
<ItemGroup>
<Reference Include="MediatR">
<HintPath>..\..\..\..\.nuget\packages\mediatr\12.5.0\lib\net6.0\MediatR.dll</HintPath>

62
Femto.Modules.Media/Infrastructure/FilesystemStorageProvider.cs Normal file
View file

@ -0,0 +1,62 @@
using Femto.Modules.Media.Data;
namespace Femto.Modules.Media.Infrastructure;
internal class FilesystemStorageProvider : IStorageProvider
{
private readonly string _storageRoot;
public FilesystemStorageProvider(string storageRoot)
{
if (string.IsNullOrWhiteSpace(storageRoot))
throw new ArgumentException(
"Storage root cannot be null or empty.",
nameof(storageRoot)
);
this._storageRoot = Path.GetFullPath(storageRoot);
if (!Directory.Exists(this._storageRoot))
{
Directory.CreateDirectory(this._storageRoot);
}
}
public async Task SaveBlob(string id, Stream data)
{
string filePath = this.GetSafeFilePath(id);
// Ensure the directory exists
Directory.CreateDirectory(Path.GetDirectoryName(filePath)!);
await using var fileStream = new FileStream(filePath, FileMode.Create, FileAccess.Write);
await data.CopyToAsync(fileStream);
}
public async Task<Stream> LoadBlob(string id)
{
string filePath = this.GetSafeFilePath(id);
if (!File.Exists(filePath))
throw new FileNotFoundException("The blob with the specified ID was not found.", id);
Stream fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read);
return fileStream;
}
private string GetSafeFilePath(string id)
{
if (string.IsNullOrWhiteSpace(id))
throw new ArgumentException("Blob ID cannot be null or empty.", nameof(id));
// Sanitize and validate the path
string combinedPath = Path.Combine(this._storageRoot, id);
string fullPath = Path.GetFullPath(combinedPath);
if (!fullPath.StartsWith(this._storageRoot, StringComparison.Ordinal))
throw new UnauthorizedAccessException("Access to the path is denied.");
return fullPath;
}
}

7
Femto.Modules.Media/Infrastructure/IStorageProvider.cs Normal file
View file

@ -0,0 +1,7 @@
namespace Femto.Modules.Media.Infrastructure;
internal interface IStorageProvider
{
Task SaveBlob(string id, Stream data);
Task<Stream> LoadBlob(string id);
}

78
Femto.Modules.Media/Infrastructure/Integration/Mailman.cs
View file

@ -1,78 +0,0 @@
using System.Text.Json;
using Femto.Modules.Media.Data;
using MediatR;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace Femto.Modules.Media.Infrastructure.Integration;
internal class Mailman(Outbox outbox, MediaContext context, ILogger<Mailman> logger, IMediator mediator)
: BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
{
var timeToWait = TimeSpan.FromSeconds(1);
while (!cancellationToken.IsCancellationRequested)
{
try
{
await this.DeliverMail(cancellationToken);
}
catch (Exception e)
{
logger.LogError(e, "Error while processing outbox");
}
try
{
await Task.Delay(timeToWait, cancellationToken);
}
catch (TaskCanceledException)
{
break;
}
}
}
private async Task DeliverMail(CancellationToken cancellationToken)
{
var messages = await outbox.GetPendingMessages(cancellationToken);
foreach (var message in messages)
{
try
{
var notificationType = OutboxMessageTypeRegistry.GetType(message.EventType);
if (notificationType is null)
{
logger.LogWarning("unmapped event type {Type}. skipping.", message.EventType);
continue;
}
var notification =
JsonSerializer.Deserialize(message.Payload, notificationType) as INotification;
if (notification is null)
throw new Exception("notification is null");
await mediator.Publish(notification, cancellationToken);
message.Succeed();
}
catch (Exception e)
{
logger.LogError(
e,
"Error processing event {EventId} for aggregate {AggregateId}",
message.Id,
message.AggregateId
);
message.Fail(e.ToString());
}
await context.SaveChangesAsync(cancellationToken);
}
}
}

34
Femto.Modules.Media/Infrastructure/Integration/Outbox.cs
View file

@ -1,34 +0,0 @@
using System.Text.Json;
using Femto.Common.Integration;
using Femto.Modules.Media.Data;
using Microsoft.EntityFrameworkCore;
namespace Femto.Modules.Media.Infrastructure.Integration;
internal class Outbox(MediaContext context)
{
public async Task AddMessage<TMessage>(Guid aggregateId, TMessage message, CancellationToken cancellationToken)
where TMessage : IIntegrationEvent
{
await context.Outbox.AddAsync(
new(
message.EventId,
aggregateId,
typeof(TMessage).Name,
JsonSerializer.Serialize(message)
),
cancellationToken
);
}
public async Task<IEnumerable<OutboxEntry>> GetPendingMessages(CancellationToken cancellationToken)
{
var now = DateTime.UtcNow;
return await context
.Outbox.Where(message => message.Status == OutboxEntryStatus.Pending)
.Where(message => message.NextRetryAt == null || message.NextRetryAt <= now)
.OrderBy(message => message.CreatedAt)
.ToListAsync(cancellationToken);
}
}

59
Femto.Modules.Media/Infrastructure/Integration/OutboxEntry.cs
View file

@ -1,59 +0,0 @@
namespace Femto.Modules.Media.Infrastructure.Integration;
internal class OutboxEntry
{
private const int MaxRetries = 5;
public Guid Id { get; private set; }
public string EventType { get; private set; } = null!;
public Guid AggregateId { get; private set; }
public string Payload { get; private set; } = null!;
public DateTime CreatedAt { get; private set; }
public DateTime? ProcessedAt { get; private set; }
public DateTime? NextRetryAt { get; private set; }
public int RetryCount { get; private set; } = 0;
public string? LastError { get; private set; }
public OutboxEntryStatus Status { get; private set; }
private OutboxEntry() { }
public OutboxEntry(Guid eventId, Guid aggregateId, string eventType, string payload)
{
this.Id = eventId;
this.EventType = eventType;
this.AggregateId = aggregateId;
this.Payload = payload;
this.CreatedAt = DateTime.UtcNow;
}
public void Succeed()
{
this.ProcessedAt = DateTime.UtcNow;
this.Status = OutboxEntryStatus.Completed;
}
public void Fail(string error)
{
if (this.RetryCount >= MaxRetries)
{
this.Status = OutboxEntryStatus.Failed;
}
else
{
this.LastError = error;
this.NextRetryAt = DateTime.UtcNow.AddSeconds(Math.Pow(2, this.RetryCount));
this.RetryCount++;
}
}
}
public enum OutboxEntryStatus
{
Pending,
Completed,
Failed
}

35
Femto.Modules.Media/Infrastructure/Integration/OutboxMessageTypeRegistry.cs
View file

@ -1,35 +0,0 @@
using System.Collections.Concurrent;
using System.Reflection;
using Femto.Common.Attributes;
using MediatR;
namespace Femto.Modules.Media.Infrastructure.Integration;
internal static class OutboxMessageTypeRegistry
{
private static readonly ConcurrentDictionary<string, Type> Mapping = new();
public static void RegisterOutboxMessageTypesInAssembly(Assembly assembly)
{
var types = assembly.GetTypes();
foreach (var type in types)
{
if (!typeof(INotification).IsAssignableFrom(type) || type.IsAbstract || type.IsInterface)
continue;
var attribute = type.GetCustomAttribute<EventTypeAttribute>();
if (attribute == null)
continue;
var eventName = attribute.Name;
if (!string.IsNullOrWhiteSpace(eventName))
{
Mapping.TryAdd(eventName, type);
}
}
}
public static Type? GetType(string eventName) => Mapping.GetValueOrDefault(eventName);
}

36
Femto.Modules.Media/Infrastructure/PipelineBehaviours/DomainEventsPipelineHandler.cs
View file

@ -1,36 +0,0 @@
using Femto.Common.Domain;
using Femto.Modules.Media.Data;
using MediatR;
namespace Femto.Modules.Media.Infrastructure.PipelineBehaviours;
internal class DomainEventsPipelineBehaviour<TRequest, TResponse>(
MediaContext context,
IPublisher publisher) : IPipelineBehavior<TRequest, TResponse>
where TRequest : notnull
{
public async Task<TResponse> Handle(
TRequest request,
RequestHandlerDelegate<TResponse> next,
CancellationToken cancellationToken)
{
var response = await next(cancellationToken);
var domainEvents = context.ChangeTracker
.Entries<Entity>()
.SelectMany(e =>
{
var events = e.Entity.DomainEvents;
e.Entity.ClearDomainEvents();
return events;
})
.ToList();
foreach (var domainEvent in domainEvents)
{
await publisher.Publish(domainEvent, cancellationToken);
}
return response;
}
}

23
Femto.Modules.Media/Infrastructure/PipelineBehaviours/SaveChangesPipelineBehaviour.cs
View file

@ -1,23 +0,0 @@
using Femto.Modules.Media.Data;
using MediatR;
namespace Femto.Modules.Media.Infrastructure.PipelineBehaviours;
/// <summary>
/// automatically call unit of work after all requuests
/// </summary>
internal class SaveChangesPipelineBehaviour<TRequest, TResponse>(MediaContext context)
: IPipelineBehavior<TRequest, TResponse> where TRequest : notnull
{
public async Task<TResponse> Handle(
TRequest request,
RequestHandlerDelegate<TResponse> next,
CancellationToken cancellationToken
)
{
var response = await next(cancellationToken);
if (context.ChangeTracker.HasChanges())
await context.SaveChangesAsync(cancellationToken);
return response;
}
}

66
Femto.Modules.Media/MediaModule.cs
View file

@ -1,66 +0,0 @@
using Femto.Modules.Media.Data;
using Femto.Modules.Media.Infrastructure.Integration;
using Femto.Modules.Media.Infrastructure.PipelineBehaviours;
using MediatR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Femto.Modules.Media;
public static class MediaModule
{
public static void UseBlogModule(this IServiceCollection services, string connectionString)
{
OutboxMessageTypeRegistry.RegisterOutboxMessageTypesInAssembly(typeof(MediaModule).Assembly);
services.AddDbContext<MediaContext>(builder =>
{
builder.UseNpgsql(
connectionString,
o =>
{
o.MapEnum<OutboxEntryStatus>("outbox_status");
}
);
;
builder.UseSnakeCaseNamingConvention();
var loggerFactory = LoggerFactory.Create(b =>
{
b.AddConsole();
// .AddFilter(
// (category, level) =>
// category == DbLoggerCategory.Database.Command.Name
// && level == LogLevel.Debug
// );
});
builder.UseLoggerFactory(loggerFactory);
builder.EnableSensitiveDataLogging();
});
services.AddMediatR(c =>
{
c.RegisterServicesFromAssembly(typeof(MediaModule).Assembly);
});
services.SetupMediatrPipeline();
services.AddTransient<Outbox, Outbox>();
services.AddHostedService<Mailman>();
}
private static void SetupMediatrPipeline(this IServiceCollection services)
{
services.AddTransient(
typeof(IPipelineBehavior<,>),
typeof(DomainEventsPipelineBehaviour<,>)
);
services.AddTransient(
typeof(IPipelineBehavior<,>),
typeof(SaveChangesPipelineBehaviour<,>)
);
}
}

22
Femto.Modules.Media/Module.cs Normal file
View file

@ -0,0 +1,22 @@
using Femto.Modules.Media.Data;
using Femto.Modules.Media.Infrastructure;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Femto.Modules.Media;
public static class Module
{
public static void UseMediaModule(this IServiceCollection services, string connectionString, string storageRoot)
{
services.AddDbContext<MediaContext>(builder =>
{
builder.UseNpgsql(connectionString);
builder.UseSnakeCaseNamingConvention();
});
services.AddTransient<IStorageProvider>(s => new FilesystemStorageProvider(storageRoot));
services.AddMediatR(c => c.RegisterServicesFromAssembly(typeof(Module).Assembly));
}
}