some changes
This commit is contained in:
parent
4ec9720541
commit
b47bac67ca
37 changed files with 397 additions and 190 deletions
|
@ -29,8 +29,4 @@
|
|||
</Reference>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Folder Include="Infrastructure\" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
32
Femto.Api/Infrastructure/EventBus.cs
Normal file
32
Femto.Api/Infrastructure/EventBus.cs
Normal file
|
@ -0,0 +1,32 @@
|
|||
using System.Threading.Channels;
|
||||
using Femto.Common.Integration;
|
||||
|
||||
namespace Femto.Api.Infrastructure;
|
||||
|
||||
public class EventBus(Channel<IEvent> channel) : BackgroundService, IEventBus
|
||||
{
|
||||
private readonly ICollection<IEventBus.Subscriber> _subscribers = [];
|
||||
|
||||
public Task Publish<T>(T evt)
|
||||
where T : IEvent
|
||||
{
|
||||
channel.Writer.TryWrite(evt);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public void Subscribe(IEventBus.Subscriber subscriber)
|
||||
{
|
||||
this._subscribers.Add(subscriber);
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await foreach (var message in channel.Reader.ReadAllAsync(stoppingToken))
|
||||
{
|
||||
await Task.WhenAll(
|
||||
this._subscribers.Select(subscriber => subscriber.Invoke(message, stoppingToken))
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
|
@ -1,9 +1,12 @@
|
|||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using System.Threading.Channels;
|
||||
using Femto.Api;
|
||||
using Femto.Api.Auth;
|
||||
using Femto.Api.Infrastructure;
|
||||
using Femto.Common;
|
||||
using Femto.Common.Domain;
|
||||
using Femto.Common.Integration;
|
||||
using Femto.Modules.Auth.Application;
|
||||
using Femto.Modules.Blog.Application;
|
||||
using Femto.Modules.Media.Application;
|
||||
|
@ -27,9 +30,12 @@ if (blobStorageRoot is null)
|
|||
throw new Exception("no blob storage root found");
|
||||
|
||||
|
||||
builder.Services.InitializeBlogModule(connectionString);
|
||||
var eventBus = new EventBus(Channel.CreateUnbounded<IEvent>());
|
||||
builder.Services.AddHostedService(_ => eventBus);
|
||||
|
||||
builder.Services.InitializeBlogModule(connectionString, eventBus);
|
||||
builder.Services.InitializeMediaModule(connectionString, blobStorageRoot);
|
||||
builder.Services.InitializeAuthenticationModule(connectionString);
|
||||
builder.Services.InitializeAuthenticationModule(connectionString, eventBus);
|
||||
|
||||
builder.Services.AddScoped<CurrentUserContext, CurrentUserContext>();
|
||||
builder.Services.AddScoped<ICurrentUserContext>(s => s.GetRequiredService<CurrentUserContext>());
|
||||
|
|
35
Femto.Common/Infrastructure/DbContextDomainExtensions.cs
Normal file
35
Femto.Common/Infrastructure/DbContextDomainExtensions.cs
Normal file
|
@ -0,0 +1,35 @@
|
|||
using Femto.Common.Domain;
|
||||
using MediatR;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Femto.Common.Infrastructure;
|
||||
|
||||
public static class DbContextDomainExtensions
|
||||
{
|
||||
public static async Task EmitDomainEvents(this DbContext context, ILogger logger, IPublisher publisher, CancellationToken cancellationToken)
|
||||
{
|
||||
var domainEvents = context
|
||||
.ChangeTracker.Entries<Entity>()
|
||||
.SelectMany(e =>
|
||||
{
|
||||
var events = e.Entity.DomainEvents;
|
||||
e.Entity.ClearDomainEvents();
|
||||
return events;
|
||||
})
|
||||
.ToList();
|
||||
|
||||
logger.LogTrace("loaded {Count} domain events", domainEvents.Count);
|
||||
|
||||
foreach (var domainEvent in domainEvents)
|
||||
{
|
||||
logger.LogTrace(
|
||||
"publishing {Type} domain event {Id}",
|
||||
domainEvent.GetType().Name,
|
||||
domainEvent.EventId
|
||||
);
|
||||
await publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
|
@ -4,5 +4,5 @@ namespace Femto.Common.Infrastructure.Outbox;
|
|||
|
||||
public interface IOutboxMessageHandler
|
||||
{
|
||||
Task Publish<TNotification>(TNotification notification, CancellationToken executionContextCancellationToken);
|
||||
Task HandleMessage<TNotification>(TNotification notification, CancellationToken cancellationToken = default);
|
||||
}
|
|
@ -13,7 +13,7 @@ public class Outbox<TContext>(TContext context, IOutboxMessageMapping mapping) w
|
|||
TMessage message,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
where TMessage : IIntegrationEvent
|
||||
where TMessage : IEvent
|
||||
{
|
||||
var eventName = mapping.GetEventName(typeof(TMessage));
|
||||
if (eventName is null)
|
||||
|
|
|
@ -3,7 +3,7 @@ namespace Femto.Common.Infrastructure.Outbox;
|
|||
public class OutboxEntry
|
||||
{
|
||||
private const int MaxRetries = 5;
|
||||
|
||||
|
||||
public Guid Id { get; private set; }
|
||||
|
||||
public string EventType { get; private set; } = null!;
|
||||
|
@ -18,7 +18,7 @@ public class OutboxEntry
|
|||
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)
|
||||
|
@ -35,7 +35,7 @@ public class OutboxEntry
|
|||
this.ProcessedAt = DateTime.UtcNow;
|
||||
this.Status = OutboxEntryStatus.Completed;
|
||||
}
|
||||
|
||||
|
||||
public void Fail(string error)
|
||||
{
|
||||
if (this.RetryCount >= MaxRetries)
|
||||
|
@ -53,7 +53,7 @@ public class OutboxEntry
|
|||
|
||||
public enum OutboxEntryStatus
|
||||
{
|
||||
Pending,
|
||||
Completed,
|
||||
Failed
|
||||
}
|
||||
Pending = 0,
|
||||
Completed = 1,
|
||||
Failed = 2,
|
||||
}
|
||||
|
|
|
@ -56,7 +56,7 @@ public class OutboxProcessor<TContext>(
|
|||
message.AggregateId
|
||||
);
|
||||
|
||||
await handler.Publish(notification, executionContext.CancellationToken);
|
||||
await handler.HandleMessage(notification, executionContext.CancellationToken);
|
||||
|
||||
message.Succeed();
|
||||
}
|
||||
|
|
|
@ -9,21 +9,23 @@ namespace Femto.Common.Infrastructure.Outbox;
|
|||
|
||||
public static class OutboxServiceExtension
|
||||
{
|
||||
public static void AddOutbox<TContext>(
|
||||
public static void AddOutbox<TContext, TMessageHandler>(
|
||||
this IServiceCollection services,
|
||||
Func<IServiceProvider, TContext>? contextFactory = null
|
||||
)
|
||||
where TContext : DbContext, IOutboxContext
|
||||
where TMessageHandler : class, IOutboxMessageHandler
|
||||
{
|
||||
|
||||
services.AddSingleton<IOutboxMessageMapping, ClrTypenameMessageMapping>();
|
||||
|
||||
|
||||
services.AddScoped<IOutboxContext>(c =>
|
||||
contextFactory?.Invoke(c) ?? c.GetRequiredService<TContext>()
|
||||
);
|
||||
|
||||
services.AddScoped<Outbox<TContext>>();
|
||||
|
||||
services.AddScoped<IOutboxMessageHandler, TMessageHandler>();
|
||||
|
||||
services.AddQuartz(q =>
|
||||
{
|
||||
var jobKey = JobKey.Create(nameof(OutboxProcessor<TContext>));
|
||||
|
|
|
@ -19,39 +19,12 @@ public class SaveChangesPipelineBehaviour<TRequest, TResponse>(
|
|||
)
|
||||
{
|
||||
var response = await next(cancellationToken);
|
||||
|
||||
if (context.ChangeTracker.HasChanges())
|
||||
{
|
||||
await this.EmitDomainEvents(cancellationToken);
|
||||
|
||||
await context.EmitDomainEvents(logger, publisher, cancellationToken);
|
||||
logger.LogDebug("saving changes");
|
||||
await context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
private async Task EmitDomainEvents(CancellationToken cancellationToken)
|
||||
{
|
||||
var domainEvents = context
|
||||
.ChangeTracker.Entries<Entity>()
|
||||
.SelectMany(e =>
|
||||
{
|
||||
var events = e.Entity.DomainEvents;
|
||||
e.Entity.ClearDomainEvents();
|
||||
return events;
|
||||
})
|
||||
.ToList();
|
||||
|
||||
logger.LogTrace("loaded {Count} domain events", domainEvents.Count);
|
||||
|
||||
foreach (var domainEvent in domainEvents)
|
||||
{
|
||||
logger.LogTrace(
|
||||
"publishing {Type} domain event {Id}",
|
||||
domainEvent.GetType().Name,
|
||||
domainEvent.EventId
|
||||
);
|
||||
await publisher.Publish(domainEvent, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
6
Femto.Common/Integration/Event.cs
Normal file
6
Femto.Common/Integration/Event.cs
Normal file
|
@ -0,0 +1,6 @@
|
|||
namespace Femto.Common.Integration;
|
||||
|
||||
public abstract record Event : IEvent
|
||||
{
|
||||
public Guid EventId { get; } = Guid.CreateVersion7();
|
||||
}
|
|
@ -2,7 +2,7 @@ using MediatR;
|
|||
|
||||
namespace Femto.Common.Integration;
|
||||
|
||||
public interface IIntegrationEvent : INotification
|
||||
public interface IEvent : INotification
|
||||
{
|
||||
public Guid EventId { get; }
|
||||
}
|
13
Femto.Common/Integration/IEventBus.cs
Normal file
13
Femto.Common/Integration/IEventBus.cs
Normal file
|
@ -0,0 +1,13 @@
|
|||
namespace Femto.Common.Integration;
|
||||
|
||||
public interface IEventBus : IEventPublisher
|
||||
{
|
||||
public delegate Task Subscriber(IEvent evt, CancellationToken cancellationToken);
|
||||
void Subscribe(Subscriber subscriber);
|
||||
|
||||
}
|
||||
|
||||
public interface IEventPublisher
|
||||
{
|
||||
Task Publish<T>(T evt) where T : IEvent;
|
||||
}
|
24
Femto.Common/Integration/IEventHandler.cs
Normal file
24
Femto.Common/Integration/IEventHandler.cs
Normal file
|
@ -0,0 +1,24 @@
|
|||
namespace Femto.Common.Integration;
|
||||
|
||||
public interface IEventHandler
|
||||
{
|
||||
Task Handle(IEvent evt, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
public abstract class EventHandler<T> : IEventHandler
|
||||
where T : IEvent
|
||||
{
|
||||
protected abstract Task Handle(T evt, CancellationToken cancellationToken);
|
||||
|
||||
public async Task Handle(IEvent evt, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (evt is not T typedEvt)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"Event {evt.GetType()} is not of type {typeof(T)}"
|
||||
);
|
||||
}
|
||||
|
||||
await Handle(typedEvt, cancellationToken);
|
||||
}
|
||||
}
|
|
@ -1,6 +0,0 @@
|
|||
namespace Femto.Common.Integration;
|
||||
|
||||
public interface IIntegrationEventBus
|
||||
{
|
||||
void Subscribe<T>() where T : IIntegrationEvent;
|
||||
}
|
|
@ -1,74 +0,0 @@
|
|||
-- Migration: Init
|
||||
-- Created at: 25/04/2025 12:14:59
|
||||
|
||||
CREATE SCHEMA blog;
|
||||
|
||||
CREATE TABLE blog.author
|
||||
(
|
||||
id uuid PRIMARY KEY,
|
||||
username varchar(64) UNIQUE NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE blog.post
|
||||
(
|
||||
id uuid PRIMARY KEY,
|
||||
content text NOT NULL,
|
||||
posted_on timestamptz NOT NULL DEFAULT now(),
|
||||
author_id uuid NOT NULL REFERENCES blog.author (id) on DELETE CASCADE
|
||||
);
|
||||
|
||||
CREATE TABLE blog.post_media
|
||||
(
|
||||
id uuid PRIMARY KEY,
|
||||
post_id uuid NOT NULL REFERENCES blog.post (id) ON DELETE CASCADE,
|
||||
url text NOT NULL,
|
||||
type varchar(64),
|
||||
width int,
|
||||
height int,
|
||||
ordering int NOT NULL
|
||||
);
|
||||
|
||||
CREATE TYPE outbox_status AS ENUM ('pending', 'completed', 'failed');
|
||||
|
||||
CREATE TABLE blog.outbox
|
||||
(
|
||||
|
||||
id uuid PRIMARY KEY,
|
||||
event_type text NOT NULL,
|
||||
aggregate_id uuid NOT NULL,
|
||||
payload jsonb NOT NULL,
|
||||
created_at timestamp DEFAULT now() NOT NULL,
|
||||
processed_at timestamp,
|
||||
next_retry_at timestamp,
|
||||
retry_count int DEFAULT 0 NOT NULL,
|
||||
last_error text,
|
||||
status outbox_status DEFAULT 'pending' NOT NULL
|
||||
);
|
||||
|
||||
CREATE SCHEMA media;
|
||||
|
||||
CREATE TABLE media.saved_blob
|
||||
(
|
||||
id uuid PRIMARY KEY,
|
||||
uploaded_on timestamp DEFAULT now() NOT NULL,
|
||||
type varchar(64) NOT NULL,
|
||||
size int
|
||||
);
|
||||
|
||||
CREATE SCHEMA authn;
|
||||
|
||||
CREATE TABLE authn.user_identity
|
||||
(
|
||||
id uuid PRIMARY KEY,
|
||||
username text NOT NULL UNIQUE,
|
||||
|
||||
password_hash bytea,
|
||||
password_salt bytea
|
||||
);
|
||||
|
||||
CREATE TABLE authn.user_session
|
||||
(
|
||||
id varchar(256) PRIMARY KEY,
|
||||
user_id uuid NOT NULL REFERENCES authn.user_identity (id) ON DELETE CASCADE,
|
||||
expires timestamptz NOT NULL
|
||||
);
|
|
@ -0,0 +1,5 @@
|
|||
using Femto.Common.Integration;
|
||||
|
||||
namespace Femto.Modules.Auth.Contracts.Events;
|
||||
|
||||
public record UserWasCreatedIntegrationEvent(Guid UserId, string Username) : Event;
|
|
@ -0,0 +1,13 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net9.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Femto.Common\Femto.Common.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
|
@ -1,34 +1,44 @@
|
|||
using Femto.Common.Infrastructure;
|
||||
using Femto.Common.Infrastructure.Outbox;
|
||||
using Femto.Common.Integration;
|
||||
using Femto.Modules.Auth.Data;
|
||||
using Femto.Modules.Auth.Infrastructure;
|
||||
using MediatR;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Quartz;
|
||||
|
||||
namespace Femto.Modules.Auth.Application;
|
||||
|
||||
public static class AuthStartup
|
||||
{
|
||||
public static void InitializeAuthenticationModule(
|
||||
this IServiceCollection rootContainer,
|
||||
string connectionString
|
||||
)
|
||||
public static void InitializeAuthenticationModule(this IServiceCollection rootContainer,
|
||||
string connectionString, IEventBus eventBus)
|
||||
{
|
||||
var hostBuilder = Host.CreateDefaultBuilder();
|
||||
hostBuilder.ConfigureServices(services => ConfigureServices(services, connectionString));
|
||||
hostBuilder.ConfigureServices(services => ConfigureServices(services, connectionString, eventBus));
|
||||
var host = hostBuilder.Build();
|
||||
rootContainer.AddScoped<IAuthModule>(_ => new AuthModule(host));
|
||||
rootContainer.AddHostedService(services => new AuthApplication(host));
|
||||
eventBus.Subscribe((evt, cancellationToken) => EventSubscriber(evt, host.Services, cancellationToken));
|
||||
}
|
||||
|
||||
private static void ConfigureServices(IServiceCollection services, string connectionString)
|
||||
private static void ConfigureServices(IServiceCollection services, string connectionString, IEventPublisher publisher)
|
||||
{
|
||||
services.AddDbContext<AuthContext>(builder =>
|
||||
{
|
||||
builder.UseNpgsql(connectionString);
|
||||
builder.UseSnakeCaseNamingConvention();
|
||||
});
|
||||
|
||||
services.AddQuartzHostedService(options =>
|
||||
{
|
||||
options.WaitForJobsToComplete = true;
|
||||
});
|
||||
|
||||
services.AddOutbox<AuthContext, OutboxMessageHandler>();
|
||||
|
||||
services.AddMediatR(c => c.RegisterServicesFromAssembly(typeof(AuthStartup).Assembly));
|
||||
|
||||
|
@ -41,9 +51,35 @@ public static class AuthStartup
|
|||
|
||||
services.ConfigureDomainServices<AuthContext>();
|
||||
|
||||
services.AddMediatR(c =>
|
||||
services.AddSingleton(publisher);
|
||||
}
|
||||
|
||||
private static async Task EventSubscriber(
|
||||
IEvent evt,
|
||||
IServiceProvider provider,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
using var scope = provider.CreateScope();
|
||||
|
||||
var context = scope.ServiceProvider.GetRequiredService<AuthContext>();
|
||||
var logger = scope.ServiceProvider.GetRequiredService<ILogger<AuthApplication>>();
|
||||
var publisher = scope.ServiceProvider.GetRequiredService<IPublisher>();
|
||||
|
||||
IEventHandler? handler = evt switch
|
||||
{
|
||||
c.RegisterServicesFromAssembly(typeof(AuthStartup).Assembly);
|
||||
});
|
||||
_ => null,
|
||||
};
|
||||
|
||||
if (handler is null)
|
||||
return;
|
||||
|
||||
await handler.Handle(evt, cancellationToken);
|
||||
|
||||
if (context.ChangeTracker.HasChanges())
|
||||
{
|
||||
await context.EmitDomainEvents(logger, publisher, cancellationToken);
|
||||
await context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,23 +0,0 @@
|
|||
using Femto.Modules.Auth.Data;
|
||||
using Femto.Modules.Auth.Models;
|
||||
|
||||
namespace Femto.Modules.Auth.Contracts;
|
||||
|
||||
internal class AuthenticationService(AuthContext context) : IAuthenticationService
|
||||
{
|
||||
public async Task<UserInfo> Register(string username, string password)
|
||||
{
|
||||
var user = new UserIdentity(username).WithPassword(password);
|
||||
await context.AddAsync(user);
|
||||
await context.SaveChangesAsync();
|
||||
|
||||
return new(user.Id, user.Username);
|
||||
}
|
||||
|
||||
public async Task<UserInfo> Authenticate(string username, string password)
|
||||
{
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
}
|
||||
|
||||
public class AuthenticationError(string message, Exception inner) : Exception(message, inner);
|
|
@ -1,7 +0,0 @@
|
|||
namespace Femto.Modules.Auth.Contracts;
|
||||
|
||||
public interface IAuthenticationService
|
||||
{
|
||||
public Task<UserInfo?> Register(string username, string password);
|
||||
public Task<UserInfo?> Authenticate(string username, string password);
|
||||
}
|
|
@ -1,3 +0,0 @@
|
|||
namespace Femto.Modules.Auth.Contracts;
|
||||
|
||||
public record UserInfo(Guid UserId, string Username);
|
|
@ -14,14 +14,12 @@
|
|||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="9.0.4" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.4" />
|
||||
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4" />
|
||||
<PackageReference Include="Quartz.Extensions.Hosting" Version="3.14.0" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Femto.Common\Femto.Common.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Folder Include="Models\DomainEventHandlers\" />
|
||||
<ProjectReference Include="..\Femto.Modules.Auth.Contracts\Femto.Modules.Auth.Contracts.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
22
Femto.Modules.Auth/Infrastructure/OutboxMessageHandler.cs
Normal file
22
Femto.Modules.Auth/Infrastructure/OutboxMessageHandler.cs
Normal file
|
@ -0,0 +1,22 @@
|
|||
using Femto.Common.Infrastructure.Outbox;
|
||||
using Femto.Common.Integration;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Femto.Modules.Auth.Infrastructure;
|
||||
|
||||
public class OutboxMessageHandler(IEventPublisher publisher, ILogger<OutboxMessageHandler> logger) : IOutboxMessageHandler
|
||||
{
|
||||
public async Task HandleMessage<TNotification>(
|
||||
TNotification notification,
|
||||
CancellationToken executionContextCancellationToken
|
||||
)
|
||||
{
|
||||
if (notification is IEvent evt)
|
||||
{
|
||||
await publisher.Publish(evt);
|
||||
} else
|
||||
{
|
||||
logger.LogWarning("ignoring non IEvent {Type} in outbox message handler", typeof(TNotification));
|
||||
}
|
||||
}
|
||||
}
|
|
@ -0,0 +1,20 @@
|
|||
using Femto.Common.Infrastructure.Outbox;
|
||||
using Femto.Modules.Auth.Contracts.Events;
|
||||
using Femto.Modules.Auth.Data;
|
||||
using Femto.Modules.Auth.Models.Events;
|
||||
using MediatR;
|
||||
|
||||
namespace Femto.Modules.Auth.Models.DomainEventHandlers;
|
||||
|
||||
internal class UserWasCreatedHandler(Outbox<AuthContext> outbox)
|
||||
: INotificationHandler<UserWasCreatedEvent>
|
||||
{
|
||||
public async Task Handle(UserWasCreatedEvent notification, CancellationToken cancellationToken)
|
||||
{
|
||||
await outbox.AddMessage(
|
||||
notification.User.Id,
|
||||
new UserWasCreatedIntegrationEvent(notification.User.Id, notification.User.Username),
|
||||
cancellationToken
|
||||
);
|
||||
}
|
||||
}
|
|
@ -1,12 +1,11 @@
|
|||
namespace Femto.Modules.Auth.Models;
|
||||
|
||||
public class UserSession
|
||||
internal class UserSession
|
||||
{
|
||||
private static TimeSpan SessionTimeout { get; } = TimeSpan.FromMinutes(30);
|
||||
private static TimeSpan ExpiryBuffer { get; } = TimeSpan.FromMinutes(5);
|
||||
public string Id { get; private set; }
|
||||
public DateTimeOffset Expires { get; private set; }
|
||||
|
||||
public bool ExpiresSoon => Expires < DateTimeOffset.UtcNow + ExpiryBuffer;
|
||||
|
||||
private UserSession() {}
|
||||
|
|
|
@ -1,4 +1,5 @@
|
|||
using Femto.Common.Infrastructure.Outbox;
|
||||
using Femto.Modules.Blog.Domain.Authors;
|
||||
using Femto.Modules.Blog.Domain.Posts;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
|
@ -7,6 +8,7 @@ namespace Femto.Modules.Blog.Application;
|
|||
internal class BlogContext(DbContextOptions<BlogContext> options) : DbContext(options), IOutboxContext
|
||||
{
|
||||
public virtual DbSet<Post> Posts { get; set; }
|
||||
public virtual DbSet<Author> Authors { get; set; }
|
||||
public virtual DbSet<OutboxEntry> Outbox { get; set; }
|
||||
|
||||
protected override void OnModelCreating(ModelBuilder builder)
|
||||
|
|
|
@ -1,6 +1,11 @@
|
|||
using Femto.Common.Infrastructure;
|
||||
using System.Runtime.CompilerServices;
|
||||
using Femto.Common.Infrastructure;
|
||||
using Femto.Common.Infrastructure.DbConnection;
|
||||
using Femto.Common.Infrastructure.Outbox;
|
||||
using Femto.Common.Integration;
|
||||
using Femto.Modules.Auth.Contracts.Events;
|
||||
using Femto.Modules.Blog.Handlers;
|
||||
using Femto.Modules.Blog.Infrastructure;
|
||||
using MediatR;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
@ -14,21 +19,32 @@ public static class BlogStartup
|
|||
{
|
||||
public static void InitializeBlogModule(
|
||||
this IServiceCollection rootContainer,
|
||||
string connectionString
|
||||
string connectionString,
|
||||
IEventBus bus
|
||||
)
|
||||
{
|
||||
var hostBuilder = Host.CreateDefaultBuilder();
|
||||
|
||||
hostBuilder.ConfigureServices(services => ConfigureServices(services, connectionString));
|
||||
hostBuilder.ConfigureServices(services =>
|
||||
ConfigureServices(services, connectionString, bus)
|
||||
);
|
||||
|
||||
var host = hostBuilder.Build();
|
||||
|
||||
rootContainer.AddHostedService(services => new BlogApplication(host));
|
||||
|
||||
rootContainer.AddScoped<IBlogModule>(_ => new BlogModule(host));
|
||||
|
||||
bus.Subscribe(
|
||||
(evt, cancellationToken) => EventSubscriber(evt, host.Services, cancellationToken)
|
||||
);
|
||||
}
|
||||
|
||||
private static void ConfigureServices(this IServiceCollection services, string connectionString)
|
||||
private static void ConfigureServices(
|
||||
this IServiceCollection services,
|
||||
string connectionString,
|
||||
IEventPublisher publisher
|
||||
)
|
||||
{
|
||||
services.AddTransient<IDbConnectionFactory>(_ => new DbConnectionFactory(connectionString));
|
||||
|
||||
|
@ -46,6 +62,8 @@ public static class BlogStartup
|
|||
builder.UseLoggerFactory(loggerFactory);
|
||||
builder.EnableSensitiveDataLogging();
|
||||
});
|
||||
|
||||
services.AddOutbox<BlogContext, OutboxMessageHandler>();
|
||||
|
||||
services.AddMediatR(c =>
|
||||
{
|
||||
|
@ -53,5 +71,42 @@ public static class BlogStartup
|
|||
});
|
||||
|
||||
services.ConfigureDomainServices<BlogContext>();
|
||||
|
||||
services.AddSingleton(publisher);
|
||||
}
|
||||
|
||||
private static async Task EventSubscriber(
|
||||
IEvent evt,
|
||||
IServiceProvider provider,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
using var scope = provider.CreateScope();
|
||||
|
||||
var context = scope.ServiceProvider.GetRequiredService<BlogContext>();
|
||||
var loggerFactory = scope.ServiceProvider.GetRequiredService<ILoggerFactory>();
|
||||
var logger = loggerFactory.CreateLogger<BlogApplication>();
|
||||
var publisher = scope.ServiceProvider.GetRequiredService<IPublisher>();
|
||||
|
||||
// todo inject these
|
||||
IEventHandler? handler = evt switch
|
||||
{
|
||||
UserWasCreatedIntegrationEvent => new UserCreatedEventHandler(
|
||||
context,
|
||||
loggerFactory.CreateLogger<UserCreatedEventHandler>()
|
||||
),
|
||||
_ => null,
|
||||
};
|
||||
|
||||
if (handler is null)
|
||||
return;
|
||||
|
||||
await handler.Handle(evt, cancellationToken);
|
||||
|
||||
if (context.ChangeTracker.HasChanges())
|
||||
{
|
||||
await context.EmitDomainEvents(logger, publisher, cancellationToken);
|
||||
await context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,14 @@
|
|||
using Femto.Modules.Blog.Domain.Authors;
|
||||
using Femto.Modules.Blog.Domain.Posts;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.EntityFrameworkCore.Metadata.Builders;
|
||||
|
||||
namespace Femto.Modules.Blog.Application.Configurations;
|
||||
|
||||
internal class AuthorConfiguration : IEntityTypeConfiguration<Author>
|
||||
{
|
||||
public void Configure(EntityTypeBuilder<Author> table)
|
||||
{
|
||||
table.ToTable("author");
|
||||
}
|
||||
}
|
|
@ -9,7 +9,5 @@ internal class OutboxEntryConfiguration : IEntityTypeConfiguration<OutboxEntry>
|
|||
public void Configure(EntityTypeBuilder<OutboxEntry> builder)
|
||||
{
|
||||
builder.ToTable("outbox");
|
||||
|
||||
builder.Property(x => x.Payload).HasColumnType("jsonb");
|
||||
}
|
||||
}
|
||||
|
|
17
Femto.Modules.Blog/Domain/Authors/Author.cs
Normal file
17
Femto.Modules.Blog/Domain/Authors/Author.cs
Normal file
|
@ -0,0 +1,17 @@
|
|||
using Femto.Common.Domain;
|
||||
|
||||
namespace Femto.Modules.Blog.Domain.Authors;
|
||||
|
||||
public class Author : Entity
|
||||
{
|
||||
public Guid Id { get; private set; }
|
||||
public string Username { get; private set; } = null!;
|
||||
|
||||
private Author() { }
|
||||
|
||||
public Author(Guid userId, string username)
|
||||
{
|
||||
this.Id = userId;
|
||||
this.Username = username;
|
||||
}
|
||||
}
|
|
@ -4,5 +4,5 @@ using Femto.Common.Integration;
|
|||
namespace Femto.Modules.Blog.Events;
|
||||
|
||||
[EventType("post.created")]
|
||||
public record PostCreatedIntegrationEvent(Guid EventId, Guid PostId, IEnumerable<Guid> MediaIds)
|
||||
: IIntegrationEvent;
|
||||
public record PostCreatedEvent(Guid EventId, Guid PostId, IEnumerable<Guid> MediaIds)
|
||||
: IEvent;
|
|
@ -38,12 +38,9 @@
|
|||
</Reference>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Folder Include="Domain\" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Femto.Common\Femto.Common.csproj" />
|
||||
<ProjectReference Include="..\Femto.Modules.Auth.Contracts\Femto.Modules.Auth.Contracts.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
|
@ -3,9 +3,9 @@ using MediatR;
|
|||
|
||||
namespace Femto.Modules.Blog.Handlers;
|
||||
|
||||
public class PostCreatedIntegrationEventHandler : INotificationHandler<PostCreatedIntegrationEvent>
|
||||
public class PostCreatedIntegrationEventHandler : INotificationHandler<PostCreatedEvent>
|
||||
{
|
||||
public async Task Handle(PostCreatedIntegrationEvent notification, CancellationToken cancellationToken)
|
||||
public async Task Handle(PostCreatedEvent notification, CancellationToken cancellationToken)
|
||||
{
|
||||
// todo
|
||||
}
|
||||
|
|
29
Femto.Modules.Blog/Handlers/UserCreatedEventHandler.cs
Normal file
29
Femto.Modules.Blog/Handlers/UserCreatedEventHandler.cs
Normal file
|
@ -0,0 +1,29 @@
|
|||
using Femto.Modules.Auth.Contracts.Events;
|
||||
using Femto.Modules.Blog.Application;
|
||||
using Femto.Modules.Blog.Domain.Authors;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Femto.Modules.Blog.Handlers;
|
||||
|
||||
internal class UserCreatedEventHandler(BlogContext context, ILogger<UserCreatedEventHandler> logger) : Common.Integration.EventHandler<UserWasCreatedIntegrationEvent>
|
||||
{
|
||||
protected override async Task Handle(UserWasCreatedIntegrationEvent evt, CancellationToken cancellationToken)
|
||||
{
|
||||
if (await context.Authors.AnyAsync(x => x.Username == evt.Username, cancellationToken))
|
||||
{
|
||||
logger.LogError("can't create author: author with username {Username} already exists", evt.Username);
|
||||
return;
|
||||
}
|
||||
|
||||
if (await context.Authors.AnyAsync(x => x.Id == evt.UserId, cancellationToken))
|
||||
{
|
||||
logger.LogError("can't create author: author with id {UserId} already exists", evt.UserId);
|
||||
return;
|
||||
}
|
||||
|
||||
var author = new Author(evt.UserId, evt.Username);
|
||||
|
||||
await context.Authors.AddAsync(author, cancellationToken);
|
||||
}
|
||||
}
|
22
Femto.Modules.Blog/Infrastructure/OutboxMessageHandler.cs
Normal file
22
Femto.Modules.Blog/Infrastructure/OutboxMessageHandler.cs
Normal file
|
@ -0,0 +1,22 @@
|
|||
using Femto.Common.Infrastructure.Outbox;
|
||||
using Femto.Common.Integration;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Femto.Modules.Blog.Infrastructure;
|
||||
|
||||
public class OutboxMessageHandler(IEventPublisher publisher, ILogger<OutboxMessageHandler> logger) : IOutboxMessageHandler
|
||||
{
|
||||
public async Task HandleMessage<TNotification>(
|
||||
TNotification notification,
|
||||
CancellationToken executionContextCancellationToken
|
||||
)
|
||||
{
|
||||
if (notification is IEvent evt)
|
||||
{
|
||||
await publisher.Publish(evt);
|
||||
} else
|
||||
{
|
||||
logger.LogWarning("ignoring non IEvent {Type} in outbox message handler", typeof(TNotification));
|
||||
}
|
||||
}
|
||||
}
|
|
@ -14,6 +14,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Femto.Modules.Media", "Femt
|
|||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Femto.Modules.Auth", "Femto.Modules.Auth\Femto.Modules.Auth.csproj", "{7E138EF6-E075-4896-93C0-923024F0CA78}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Femto.Modules.Auth.Contracts", "Femto.Modules.Auth.Contracts\Femto.Modules.Auth.Contracts.csproj", "{1AC1DA1D-54B0-44FC-9FDF-9C2E68BB8ABB}"
|
||||
EndProject
|
||||
Global
|
||||
GlobalSection(SolutionConfigurationPlatforms) = preSolution
|
||||
Debug|Any CPU = Debug|Any CPU
|
||||
|
@ -48,6 +50,10 @@ Global
|
|||
{7E138EF6-E075-4896-93C0-923024F0CA78}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{7E138EF6-E075-4896-93C0-923024F0CA78}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{7E138EF6-E075-4896-93C0-923024F0CA78}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{1AC1DA1D-54B0-44FC-9FDF-9C2E68BB8ABB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{1AC1DA1D-54B0-44FC-9FDF-9C2E68BB8ABB}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{1AC1DA1D-54B0-44FC-9FDF-9C2E68BB8ABB}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{1AC1DA1D-54B0-44FC-9FDF-9C2E68BB8ABB}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
EndGlobalSection
|
||||
GlobalSection(NestedProjects) = preSolution
|
||||
EndGlobalSection
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue