conco/FictionArchive
1
0
Fork 0
Code Issues 13 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

[FA-misc] Mass transit overhaul, needs testing and review

Browse Source
This commit is contained in:
gamer147
2026-01-21 23:16:31 -05:00
parent 055ef33666
commit f88f340d0a
97 changed files with 1150 additions and 858 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
FictionArchive.Service.Shared/FictionArchive.Service.Shared.csproj
View File

@@ -31,7 +31,9 @@
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4" />
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL.NodaTime" Version="9.0.4" />
<PackageReference Include="Polly" Version="8.6.5" />
<PackageReference Include="RabbitMQ.Client" Version="7.2.0" />
<PackageReference Include="MassTransit" Version="8.4.0" />
<PackageReference Include="MassTransit.RabbitMQ" Version="8.4.0" />
<PackageReference Include="MassTransit.EntityFrameworkCore" Version="8.4.0" />
<PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="8.0.11" />
</ItemGroup>

9
FictionArchive.Service.Shared/MassTransit/Configuration/MassTransitOptions.cs Normal file
View File

@@ -0,0 +1,9 @@
namespace FictionArchive.Service.Shared.MassTransit.Configuration;
public class MassTransitOptions
{
public string Host { get; set; } = "localhost";
public string VirtualHost { get; set; } = "/";
public string Username { get; set; } = "guest";
public string Password { get; set; } = "guest";
}

6
FictionArchive.Service.Shared/MassTransit/Contracts/Commands/ImportNovelCommand.cs Normal file
View File

@@ -0,0 +1,6 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Commands;
public record ImportNovelCommand : ICommand
{
public required string NovelUrl { get; init; }
}

8
FictionArchive.Service.Shared/MassTransit/Contracts/Commands/PullChapterContentCommand.cs Normal file
View File

@@ -0,0 +1,8 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Commands;
public record PullChapterContentCommand : ICommand
{
public required uint NovelId { get; init; }
public required uint VolumeId { get; init; }
public required uint ChapterOrder { get; init; }
}

12
FictionArchive.Service.Shared/MassTransit/Contracts/Commands/TranslateTextCommand.cs Normal file
View File

@@ -0,0 +1,12 @@
using FictionArchive.Common.Enums;
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Commands;
public record TranslateTextCommand : ICommand
{
public Guid TranslationRequestId { get; init; }
public Language From { get; init; }
public Language To { get; init; }
public required string Body { get; init; }
public required string TranslationEngineKey { get; init; }
}

8
FictionArchive.Service.Shared/MassTransit/Contracts/Commands/UploadFileCommand.cs Normal file
View File

@@ -0,0 +1,8 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Commands;
public record UploadFileCommand : ICommand
{
public Guid RequestId { get; init; }
public required string FilePath { get; init; }
public required byte[] FileData { get; init; }
}

9
FictionArchive.Service.Shared/MassTransit/Contracts/Events/AuthUserAddedEvent.cs Normal file
View File

@@ -0,0 +1,9 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Events;
public record AuthUserAddedEvent : IEvent
{
public required string OAuthProviderId { get; init; }
public required string InviterOAuthProviderId { get; init; }
public required string EventUserEmail { get; init; }
public required string EventUserUsername { get; init; }
}

11
FictionArchive.Service.Shared/MassTransit/Contracts/Events/ChapterCreatedEvent.cs Normal file
View File

@@ -0,0 +1,11 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Events;
public record ChapterCreatedEvent : IEvent
{
public required uint ChapterId { get; init; }
public required uint NovelId { get; init; }
public required uint VolumeId { get; init; }
public required int VolumeOrder { get; init; }
public required uint ChapterOrder { get; init; }
public required string ChapterTitle { get; init; }
}

11
FictionArchive.Service.Shared/MassTransit/Contracts/Events/FileUploadCompletedEvent.cs Normal file
View File

@@ -0,0 +1,11 @@
using FictionArchive.Common.Enums;
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Events;
public record FileUploadCompletedEvent : IEvent
{
public Guid RequestId { get; init; }
public RequestStatus Status { get; init; }
public string? FileAccessUrl { get; init; }
public string? ErrorMessage { get; init; }
}

12
FictionArchive.Service.Shared/MassTransit/Contracts/Events/NovelCreatedEvent.cs Normal file
View File

@@ -0,0 +1,12 @@
using FictionArchive.Common.Enums;
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Events;
public record NovelCreatedEvent : IEvent
{
public required uint NovelId { get; init; }
public required string Title { get; init; }
public required Language OriginalLanguage { get; init; }
public required string Source { get; init; }
public required string AuthorName { get; init; }
}

7
FictionArchive.Service.Shared/MassTransit/Contracts/Events/TranslationCompletedEvent.cs Normal file
View File

@@ -0,0 +1,7 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Events;
public record TranslationCompletedEvent : IEvent
{
public Guid TranslationRequestId { get; init; }
public required string TranslatedText { get; init; }
}

13
FictionArchive.Service.Shared/MassTransit/Contracts/Events/UserInvitedEvent.cs Normal file
View File

@@ -0,0 +1,13 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts.Events;
public record UserInvitedEvent : IEvent
{
public Guid InvitedUserId { get; init; }
public required string InvitedUsername { get; init; }
public required string InvitedEmail { get; init; }
public required string InvitedOAuthProviderId { get; init; }
public Guid InviterId { get; init; }
public required string InviterUsername { get; init; }
public required string InviterOAuthProviderId { get; init; }
}

6
FictionArchive.Service.Shared/MassTransit/Contracts/ICommand.cs Normal file
View File

@@ -0,0 +1,6 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts;
/// <summary>
/// Marker interface for commands (do something, single consumer)
/// </summary>
public interface ICommand { }

6
FictionArchive.Service.Shared/MassTransit/Contracts/IEvent.cs Normal file
View File

@@ -0,0 +1,6 @@
namespace FictionArchive.Service.Shared.MassTransit.Contracts;
/// <summary>
/// Marker interface for events (something happened, multiple subscribers)
/// </summary>
public interface IEvent { }

18
FictionArchive.Service.Shared/MassTransit/Contracts/JobStateChangedEvent.cs Normal file
View File

@@ -0,0 +1,18 @@
using NodaTime;
namespace FictionArchive.Service.Shared.MassTransit.Contracts;
/// <summary>
/// Published by sagas on state transitions for centralized job tracking
/// </summary>
public record JobStateChangedEvent : IEvent
{
public Guid JobId { get; init; }
public required string JobType { get; init; }
public required string FromState { get; init; }
public required string ToState { get; init; }
public string? Message { get; init; }
public string? Error { get; init; }
public Instant Timestamp { get; init; }
public Dictionary<string, object>? Metadata { get; init; }
}

103
FictionArchive.Service.Shared/MassTransit/MassTransitExtensions.cs Normal file
View File

@@ -0,0 +1,103 @@
using FictionArchive.Service.Shared.MassTransit.Configuration;
using MassTransit;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
namespace FictionArchive.Service.Shared.MassTransit;
public static class MassTransitExtensions
{
/// <summary>
/// Adds MassTransit with RabbitMQ and Entity Framework outbox
/// </summary>
public static IServiceCollection AddFictionArchiveMassTransit<TDbContext>(
this IServiceCollection services,
IConfiguration configuration,
Action<IBusRegistrationConfigurator>? configureConsumers = null)
where TDbContext : DbContext
{
services.AddMassTransit(x =>
{
configureConsumers?.Invoke(x);
x.AddEntityFrameworkOutbox<TDbContext>(o =>
{
o.UsePostgres();
o.UseBusOutbox();
});
x.UsingRabbitMq((context, cfg) =>
{
var options = configuration.GetSection("RabbitMQ").Get<MassTransitOptions>()
?? new MassTransitOptions();
cfg.Host(options.Host, options.VirtualHost, h =>
{
h.Username(options.Username);
h.Password(options.Password);
});
// Immediate retries for transient failures
cfg.UseMessageRetry(r => r.Intervals(
TimeSpan.FromSeconds(1),
TimeSpan.FromSeconds(1),
TimeSpan.FromSeconds(1)));
// Delayed redelivery for longer outages
cfg.UseDelayedRedelivery(r => r.Intervals(
TimeSpan.FromSeconds(5),
TimeSpan.FromSeconds(30),
TimeSpan.FromMinutes(2),
TimeSpan.FromMinutes(10),
TimeSpan.FromMinutes(30)));
cfg.ConfigureEndpoints(context);
});
});
return services;
}
/// <summary>
/// Adds MassTransit with RabbitMQ without outbox (for services without EF)
/// </summary>
public static IServiceCollection AddFictionArchiveMassTransit(
this IServiceCollection services,
IConfiguration configuration,
Action<IBusRegistrationConfigurator>? configureConsumers = null)
{
services.AddMassTransit(x =>
{
configureConsumers?.Invoke(x);
x.UsingRabbitMq((context, cfg) =>
{
var options = configuration.GetSection("RabbitMQ").Get<MassTransitOptions>()
?? new MassTransitOptions();
cfg.Host(options.Host, options.VirtualHost, h =>
{
h.Username(options.Username);
h.Password(options.Password);
});
cfg.UseMessageRetry(r => r.Intervals(
TimeSpan.FromSeconds(1),
TimeSpan.FromSeconds(1),
TimeSpan.FromSeconds(1)));
cfg.UseDelayedRedelivery(r => r.Intervals(
TimeSpan.FromSeconds(5),
TimeSpan.FromSeconds(30),
TimeSpan.FromMinutes(2),
TimeSpan.FromMinutes(10),
TimeSpan.FromMinutes(30)));
cfg.ConfigureEndpoints(context);
});
});
return services;
}
}

25
FictionArchive.Service.Shared/Services/EventBus/EventBusBuilder.cs
View File

@@ -1,25 +0,0 @@
using Microsoft.Extensions.DependencyInjection;
namespace FictionArchive.Service.Shared.Services.EventBus;
public class EventBusBuilder<TEventBus> where TEventBus : class, IEventBus
{
private readonly IServiceCollection _services;
private readonly SubscriptionManager _subscriptionManager;
public EventBusBuilder(IServiceCollection services)
{
_services = services;
_services.AddSingleton<IEventBus, TEventBus>();
_subscriptionManager = new SubscriptionManager();
_services.AddSingleton<SubscriptionManager>(_subscriptionManager);
}
public EventBusBuilder<TEventBus> Subscribe<TEvent, TEventHandler>() where TEvent : IIntegrationEvent where TEventHandler : class, IIntegrationEventHandler<TEvent>
{
_services.AddKeyedTransient<IIntegrationEventHandler, TEventHandler>(typeof(TEvent).Name);
_subscriptionManager.RegisterSubscription<TEvent>();
return this;
}
}

12
FictionArchive.Service.Shared/Services/EventBus/EventBusExtensions.cs
View File

@@ -1,12 +0,0 @@
using Microsoft.Extensions.DependencyInjection;
namespace FictionArchive.Service.Shared.Services.EventBus;
public static class EventBusExtensions
{
public static EventBusBuilder<TEventBus> AddEventBus<TEventBus>(this IServiceCollection services)
where TEventBus : class, IEventBus
{
return new EventBusBuilder<TEventBus>(services);
}
}

7
FictionArchive.Service.Shared/Services/EventBus/IEventBus.cs
View File

@@ -1,7 +0,0 @@
namespace FictionArchive.Service.Shared.Services.EventBus;
public interface IEventBus
{
Task Publish<TEvent>(TEvent integrationEvent) where TEvent : IIntegrationEvent;
Task Publish(object integrationEvent, string eventType);
}

7
FictionArchive.Service.Shared/Services/EventBus/IIntegrationEvent.cs
View File

@@ -1,7 +0,0 @@
using NodaTime;
namespace FictionArchive.Service.Shared.Services.EventBus;
public interface IIntegrationEvent
{
}

12
FictionArchive.Service.Shared/Services/EventBus/IIntegrationEventHandler.cs
View File

@@ -1,12 +0,0 @@
namespace FictionArchive.Service.Shared.Services.EventBus;
public interface IIntegrationEventHandler<in TEvent> : IIntegrationEventHandler where TEvent : IIntegrationEvent
{
Task Handle(TEvent @event);
Task IIntegrationEventHandler.Handle(IIntegrationEvent @event) => Handle((TEvent)@event);
}
public interface IIntegrationEventHandler
{
Task Handle(IIntegrationEvent @event);
}

35
FictionArchive.Service.Shared/Services/EventBus/Implementations/RabbitMQConnectionProvider.cs
View File

@@ -1,35 +0,0 @@
using RabbitMQ.Client;
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
public class RabbitMQConnectionProvider
{
private readonly IConnectionFactory _connectionFactory;
private IConnection Connection { get; set; }
private IChannel DefaultChannel { get; set; }
public RabbitMQConnectionProvider(IConnectionFactory connectionFactory)
{
_connectionFactory = connectionFactory;
}
public async Task<IConnection> GetConnectionAsync()
{
if (Connection == null)
{
Connection = await _connectionFactory.CreateConnectionAsync();
}
return Connection;
}
public async Task<IChannel> GetDefaultChannelAsync()
{
if (DefaultChannel == null)
{
DefaultChannel = await (await GetConnectionAsync()).CreateChannelAsync();
}
return DefaultChannel;
}
}

137
FictionArchive.Service.Shared/Services/EventBus/Implementations/RabbitMQEventBus.cs
View File

@@ -1,137 +0,0 @@
using System.Text;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using NodaTime;
using NodaTime.Serialization.JsonNet;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
public class RabbitMQEventBus : IEventBus, IHostedService
{
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly RabbitMQConnectionProvider _connectionProvider;
private readonly RabbitMQOptions _options;
private readonly SubscriptionManager _subscriptionManager;
private readonly ILogger<RabbitMQEventBus> _logger;
private readonly JsonSerializerSettings _jsonSerializerSettings;
private const string ExchangeName = "fiction-archive-event-bus";
private const string CreatedAtHeader = "X-Created-At";
private const string EventIdHeader = "X-Event-Id";
public RabbitMQEventBus(IServiceScopeFactory serviceScopeFactory, RabbitMQConnectionProvider connectionProvider, IOptions<RabbitMQOptions> options, SubscriptionManager subscriptionManager, ILogger<RabbitMQEventBus> logger)
{
_serviceScopeFactory = serviceScopeFactory;
_connectionProvider = connectionProvider;
_subscriptionManager = subscriptionManager;
_logger = logger;
_options = options.Value;
_jsonSerializerSettings = new JsonSerializerSettings().ConfigureForNodaTime(DateTimeZoneProviders.Tzdb);
}
public async Task Publish<TEvent>(TEvent integrationEvent) where TEvent : IIntegrationEvent
{
var routingKey = typeof(TEvent).Name;
await Publish(integrationEvent, routingKey);
}
public async Task Publish(object integrationEvent, string eventType)
{
var channel = await _connectionProvider.GetDefaultChannelAsync();
var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(integrationEvent));
// headers
var props = new BasicProperties();
props.Headers = new Dictionary<string, object?>()
{
{ CreatedAtHeader, Instant.FromDateTimeUtc(DateTime.UtcNow).ToString() },
{ EventIdHeader, Guid.NewGuid().ToString() }
};
await channel.BasicPublishAsync(ExchangeName, eventType, true, props, body);
_logger.LogInformation("Published event {EventName}", eventType);
}
public async Task StartAsync(CancellationToken cancellationToken)
{
_ = Task.Factory.StartNew(async () =>
{
try
{
var channel = await _connectionProvider.GetDefaultChannelAsync();
await channel.ExchangeDeclareAsync(ExchangeName, ExchangeType.Direct,
cancellationToken: cancellationToken);
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: cancellationToken);
await channel.QueueDeclareAsync(_options.ClientIdentifier, true, false, false,
cancellationToken: cancellationToken);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += (sender, @event) =>
{
return OnReceivedEvent(sender, @event, channel);
};
await channel.BasicConsumeAsync(_options.ClientIdentifier, false, consumer, cancellationToken: cancellationToken);
foreach (var subscription in _subscriptionManager.Subscriptions)
{
await channel.QueueBindAsync(_options.ClientIdentifier, ExchangeName, subscription.Key,
cancellationToken: cancellationToken);
_logger.LogInformation("Subscribed to {SubscriptionKey}", subscription.Key);
}
_logger.LogInformation("RabbitMQ EventBus started.");
}
catch (Exception e)
{
_logger.LogError(e, "An error occurred while starting the RabbitMQ EventBus");
}
}, cancellationToken);
}
public Task StopAsync(CancellationToken cancellationToken)
{
return Task.CompletedTask;
}
private async Task OnReceivedEvent(object sender, BasicDeliverEventArgs @event, IChannel channel)
{
var eventName = @event.RoutingKey;
_logger.LogInformation("Received event {EventName}", eventName);
try
{
if (!_subscriptionManager.Subscriptions.ContainsKey(eventName))
{
_logger.LogWarning("Received event without subscription entry.");
return;
}
var eventBody = Encoding.UTF8.GetString(@event.Body.Span);
var eventObject = JsonConvert.DeserializeObject(eventBody, _subscriptionManager.Subscriptions[eventName], _jsonSerializerSettings) as IIntegrationEvent;
using var scope = _serviceScopeFactory.CreateScope();
foreach (var service in scope.ServiceProvider.GetKeyedServices<IIntegrationEventHandler>(eventName))
{
await service.Handle(eventObject);
}
_logger.LogInformation("Finished handling event with name {EventName}", eventName);
}
catch (Exception e)
{
_logger.LogError(e, "An error occurred while handling an event.");
}
finally
{
await channel.BasicAckAsync(@event.DeliveryTag, false);
}
}
}

24
FictionArchive.Service.Shared/Services/EventBus/Implementations/RabbitMQExtensions.cs
View File

@@ -1,24 +0,0 @@
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
public static class RabbitMQExtensions
{
public static EventBusBuilder<RabbitMQEventBus> AddRabbitMQ(this IServiceCollection services, Action<RabbitMQOptions> configure)
{
services.Configure(configure);
services.AddSingleton<IConnectionFactory, ConnectionFactory>(provider =>
{
var options = provider.GetService<IOptions<RabbitMQOptions>>();
ConnectionFactory factory = new ConnectionFactory();
factory.Uri = new Uri(options.Value.ConnectionString);
return factory;
});
services.AddSingleton<RabbitMQConnectionProvider>();
services.AddHostedService<RabbitMQEventBus>();
return services.AddEventBus<RabbitMQEventBus>();
}
}

7
FictionArchive.Service.Shared/Services/EventBus/Implementations/RabbitMQOptions.cs
View File

@@ -1,7 +0,0 @@
namespace FictionArchive.Service.Shared.Services.EventBus.Implementations;
public class RabbitMQOptions
{
public string ConnectionString { get; set; }
public string ClientIdentifier { get; set; }
}

11
FictionArchive.Service.Shared/Services/EventBus/SubscriptionManager.cs
View File

@@ -1,11 +0,0 @@
namespace FictionArchive.Service.Shared.Services.EventBus;
public class SubscriptionManager
{
public Dictionary<string, Type> Subscriptions { get; } = new Dictionary<string, Type>();
public void RegisterSubscription<TEvent>()
{
Subscriptions.Add(typeof(TEvent).Name, typeof(TEvent));
}
}