diff --git a/.agents/rules/eventing.md b/.agents/rules/eventing.md index aa7cdc27d7..46243edccf 100644 --- a/.agents/rules/eventing.md +++ b/.agents/rules/eventing.md @@ -47,6 +47,8 @@ Defaults in BuildingBlocks are single-database (`SingleDatabaseDrainTargetProvid `InMemoryEventBus` resolves handlers in a fresh DI scope and applies the **Inbox**: skips if `IInboxStore.HasProcessedAsync(eventId, handlerName)`, marks processed after success. Composite key `{Id, HandlerName}`; concurrent-insert race is swallowed. Don't hand-roll dedup. +**Handlers are isolated from each other.** The bus runs every handler for an event even when an earlier one throws, then rethrows (the single exception, or an `AggregateException` for several) so the outbox retries. The inbox skips the handlers that already succeeded, so a retry re-runs only the failed ones, and a handler that keeps failing until the row dead-letters never cost the others their turn. Each handler also runs in its own DI scope, so a handler that throws cannot leave unsaved DbContext changes for the next one to persist. + ## Wiring The **host** bootstraps eventing once (`FSH.Starter.Api/Program.cs` and `FSH.Starter.DbMigrator/Program.cs`, before `AddModules` so `EventingDbInitializer` migrates the `framework` schema first): diff --git a/.agents/rules/storage.md b/.agents/rules/storage.md index 35c8d8df32..0d4bd03c50 100644 --- a/.agents/rules/storage.md +++ b/.agents/rules/storage.md @@ -13,11 +13,11 @@ Bucket policies in every stack (Aspire bootstrap, compose `rustfs-init`, Terraform `app_s3_public_read_prefix` + the CloudFront read statement) grant anonymous `s3:GetObject` on **`public/*` only** (`StorageVisibilityRoot`). So: - Files keys are `{public|private}/tenants/{tenant}/{owner}/{yyyy}/{MM}/{id:N}/{name}` — `StorageKeyBuilder` (Files module) is the **one** place that decides; never hand-build a key. - `S3StorageService.UploadAsync` always returns a public URL, so it writes under `public/uploads/…`. -- Moves go through `FileStorageRelocator.ApplyAsync`, never by hand: it takes the file's row lock (`SELECT … FOR UPDATE`, PostgreSQL), re-reads the row, then copy → commit row + `FileStorageKeyChangedIntegrationEvent` → delete old. The old object is deleted only when no row references it; the first attempt runs right after the commit, and Files' own `DeleteSupersededObjectHandler` retries it via the outbox until it is really gone (so a Public → Private flip can't leave a public copy behind). +- Moves go through `FileStorageRelocator.ApplyAsync`, never by hand: it takes the file's row lock (`SELECT … FOR UPDATE`, PostgreSQL), re-reads the row, then copy → commit row + `FileStorageKeyChangedIntegrationEvent` → delete old. If the commit fails after the copy, the copy is removed again (refunding its bytes), best-effort and logged. The old object is deleted only when no row references it; the first attempt runs right after the commit, and Files' own `DeleteSupersededObjectHandler` retries it via the outbox until it is really gone (so a Public → Private flip can't leave a public copy behind). - Modules that persist a public URL must handle that event and use `ResolveUrl` (it returns the API's current `NewPublicUrl` for public files) — Catalog (product images) and Identity (avatars) do. -- Keys from before #1410 (`tenants/…`, `uploads/…`) have no root: `MigrateLegacyPublicFileKeysJob` (enqueued via `IJobService` on every API start; keyset-paginated; retried by Hangfire) moves public FileAssets under `public/`; legacy private ones stay put. It reads the partial index `IX_FileAsset_LegacyKey`, which is empty once done, so repeat runs are nearly free. `FinalizeUpload` relocates uploads that were presigned before the upgrade. Legacy `UploadAsync` objects (`uploads/…`) are **not** migrated. +- Keys from before #1410 (`tenants/…`, `uploads/…`) have no root: `MigrateLegacyPublicFileKeysJob` (enqueued via `IJobService` on every API start; keyset-paginated; retried by Hangfire) moves public FileAssets under `public/`; legacy private ones stay put. It reads the partial index `IX_FileAsset_LegacyKey`, which is empty once done, so repeat runs are nearly free. `FinalizeUpload` relocates uploads that were presigned before the upgrade, best-effort: the finalize is already committed, so a failed move is logged and the file stays at its legacy key (the job moves a public one later; a private one stays, which is safe). Legacy `UploadAsync` objects (`uploads/…`) are **not** migrated. - `QuotaMeteredStorageService.RemoveAsync` refunds only when the object is gone afterwards (providers swallow delete errors). -- With `Storage:S3:Prefix` set, keys become `{Prefix}/public/…` — the policy must match that, and the prefix must **not** be `public` or `private`. +- With `Storage:S3:Prefix` set, keys become `{Prefix}/public/…` — the policy must match that, and the prefix's first segment must **not** be `public` or `private` (case-insensitive; `AddHeroStorage` fails startup via `ValidateOnStart`). - The **Local** provider serves everything under wwwroot: it does not enforce `private/` at all. Use it for dev only. `FileType`: `Image` (5MB), `Document`, `Pdf` (10MB) — `FileTypeMetadata.GetRules` enforces extension + size. **Always propagate `CancellationToken`.** diff --git a/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs b/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs index 7066bcda88..24afdae0c6 100644 --- a/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs +++ b/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs @@ -4,6 +4,7 @@ using Microsoft.Extensions.Logging; using System.Collections.Concurrent; using System.Reflection; +using System.Runtime.ExceptionServices; namespace FSH.Framework.Eventing.InMemory; @@ -63,23 +64,65 @@ private async Task PublishSingleAsync(IIntegrationEvent @event, CancellationToke // construction, so a late tenant NREs the query filter. This is what makes background publishers work. using (_tenantScope.Begin(@event.TenantId)) { - using var scope = _serviceProvider.CreateScope(); - var provider = scope.ServiceProvider; - - var handlers = ResolveHandlers(provider, dispatch.HandlerInterfaceType); - if (handlers.Length == 0) + // One failing handler must not starve the others: every handler gets its turn, and the failure + // surfaces afterwards so the outbox retries the event. The inbox records each handler that + // succeeded, so a retry re-runs only the ones that failed. + // + // Each handler runs in its own DI scope. A handler that throws can leave tracked but unsaved + // changes on a scoped DbContext; in a shared scope the next handler's SaveChanges would persist + // them. Handlers are resolved per scope in registration order, so index i is the same handler. + List? failures = null; + var handlerCount = -1; + for (var i = 0; handlerCount < 0 || i < handlerCount; i++) { - LogNoHandlers(eventType.FullName); - return; + using var scope = _serviceProvider.CreateScope(); + var provider = scope.ServiceProvider; + var handlers = ResolveHandlers(provider, dispatch.HandlerInterfaceType); + + if (handlerCount < 0) + { + handlerCount = handlers.Length; + if (handlerCount == 0) + { + LogNoHandlers(eventType.FullName); + return; + } + } + + var inbox = provider.GetService(); + try + { + await InvokeHandlerAsync(handlers[i], dispatch.HandleMethod, eventType, @event, inbox, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + (failures ??= []).Add(ExceptionDispatchInfo.Capture(ex)); + } } - var inbox = provider.GetService(); + ThrowIfAnyFailed(failures, @event.Id); + } + } - foreach (var handler in handlers) - { - await InvokeHandlerAsync(handler, dispatch.HandleMethod, eventType, @event, inbox, ct).ConfigureAwait(false); - } + private static void ThrowIfAnyFailed(List? failures, Guid eventId) + { + if (failures is null) + { + return; + } + + if (failures.Count == 1) + { + failures[0].Throw(); } + + throw new AggregateException( + $"{failures.Count} handlers failed for integration event {eventId}.", + failures.Select(f => f.SourceException)); } private static object[] ResolveHandlers(IServiceProvider provider, Type handlerInterfaceType) @@ -120,7 +163,8 @@ private async Task ExecuteHandlerAsync( { try { - var task = (Task)method.Invoke(handler, new object[] { @event, ct })!; + // DoNotWrapExceptions: a handler that throws synchronously surfaces its own exception, not a TargetInvocationException. + var task = (Task)method.Invoke(handler, BindingFlags.DoNotWrapExceptions, binder: null, new object[] { @event, ct }, culture: null)!; await task.ConfigureAwait(false); if (inbox != null) diff --git a/src/BuildingBlocks/Storage/Extensions.cs b/src/BuildingBlocks/Storage/Extensions.cs index 855c4a7954..ae39be889f 100644 --- a/src/BuildingBlocks/Storage/Extensions.cs +++ b/src/BuildingBlocks/Storage/Extensions.cs @@ -38,6 +38,7 @@ public static IServiceCollection AddHeroStorage(this IServiceCollection services .AddOptions() .Bind(configuration.GetSection("Storage:S3")) .Validate(o => string.IsNullOrWhiteSpace(o.PresignServiceUrl) || IsRootHttpUrl(o.PresignServiceUrl), "Storage:S3:PresignServiceUrl must be an absolute http(s) URL with no path, e.g. https://s3.example.com.") + .Validate(o => !PrefixStartsWithVisibilityRoot(o.Prefix), "Storage:S3:Prefix must not start with 'public' or 'private': keys are stored as {Prefix}/public/... and {Prefix}/private/..., and a public/* read policy would then cover private files.") .ValidateOnStart(); services.AddSingleton(sp => @@ -109,6 +110,20 @@ private static bool IsRootHttpUrl(string value) => && string.IsNullOrEmpty(uri.Query) && string.IsNullOrEmpty(uri.Fragment); + // Keys are {Prefix}/{public|private}/…. A prefix whose first segment is itself a visibility root puts + // private keys under public/ (or the other way round), where the bucket's public/* read policy applies. + private static bool PrefixStartsWithVisibilityRoot(string? prefix) + { + if (string.IsNullOrWhiteSpace(prefix)) + { + return false; + } + + var firstSegment = prefix.Trim().TrimStart('/').Split('/', 2)[0]; + return string.Equals(firstSegment + "/", StorageVisibilityRoot.Public, StringComparison.OrdinalIgnoreCase) + || string.Equals(firstSegment + "/", StorageVisibilityRoot.Private, StringComparison.OrdinalIgnoreCase); + } + private static void RegisterStorageService( IServiceCollection services, bool quotaEnabled, diff --git a/src/Modules/Files/Modules.Files/Features/v1/FinalizeUpload/FinalizeUploadCommandHandler.cs b/src/Modules/Files/Modules.Files/Features/v1/FinalizeUpload/FinalizeUploadCommandHandler.cs index 7b0ed9a85e..cbe8147c71 100644 --- a/src/Modules/Files/Modules.Files/Features/v1/FinalizeUpload/FinalizeUploadCommandHandler.cs +++ b/src/Modules/Files/Modules.Files/Features/v1/FinalizeUpload/FinalizeUploadCommandHandler.cs @@ -15,17 +15,19 @@ using FSH.Modules.Files.Services; using Mediator; using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; namespace FSH.Modules.Files.Features.v1.FinalizeUpload; -public sealed class FinalizeUploadCommandHandler( +public sealed partial class FinalizeUploadCommandHandler( FilesDbContext db, IStorageService storage, IFileScanner scanner, IQuotaService quotas, IOutboxWriter outbox, ICurrentUser currentUser, - FileStorageRelocator relocator) + FileStorageRelocator relocator, + ILogger logger) : ICommandHandler { public async ValueTask Handle(FinalizeUploadCommand cmd, CancellationToken cancellationToken) @@ -105,13 +107,34 @@ await outbox.AddAsync(new FileFinalizedIntegrationEvent( if (asset.Status == FileAssetStatus.Available && !StorageKeyBuilder.IsUnderVisibilityRoot(asset.StorageKey, asset.Visibility)) { - asset = await relocator.ApplyAsync( + asset = await TryMoveUnderVisibilityRootAsync(asset, cancellationToken).ConfigureAwait(false); + } + + return FileAssetMapper.ToDto(asset); + } + + // Best-effort: the finalize above is already committed, so a failed move must not turn into an error + // for a file that is Available (a retried finalize would only get 409). The file stays at its legacy + // key: a public one is moved by MigrateLegacyPublicFileKeysJob on its next run; a private one stays + // there, which is safe because no bucket policy grants anonymous read outside public/*. + private async Task TryMoveUnderVisibilityRootAsync(FileAsset asset, CancellationToken cancellationToken) + { + try + { + return await relocator.ApplyAsync( asset.Id, a => a.Status == FileAssetStatus.Available && !StorageKeyBuilder.IsUnderVisibilityRoot(a.StorageKey, a.Visibility), cancellationToken).ConfigureAwait(false) ?? asset; } - - return FileAssetMapper.ToDto(asset); + catch (Exception ex) when (ex is not OperationCanceledException) + { + LogMoveDeferred(logger, ex, asset.Id, asset.StorageKey); + return asset; + } } + + [LoggerMessage(Level = LogLevel.Warning, + Message = "[Files] finalized file {FileAssetId} but could not move it from legacy key {StorageKey}; it stays there until the legacy-key migration moves it")] + private static partial void LogMoveDeferred(ILogger logger, Exception exception, Guid fileAssetId, string storageKey); } diff --git a/src/Modules/Files/Modules.Files/Services/FileStorageRelocator.cs b/src/Modules/Files/Modules.Files/Services/FileStorageRelocator.cs index 8bd4526ead..0fe66ab295 100644 --- a/src/Modules/Files/Modules.Files/Services/FileStorageRelocator.cs +++ b/src/Modules/Files/Modules.Files/Services/FileStorageRelocator.cs @@ -30,6 +30,8 @@ namespace FSH.Modules.Files.Services; /// DeleteSupersededObjectHandler when the outbox delivers the event, which retries with backoff /// until the object is really gone. A crash or a swallowed store error therefore leaves at most a /// temporary orphan, never a row pointing at nothing and never a public copy of a private file for good. +/// When the copy succeeds but the commit fails, the copy is removed again (and its bytes refunded), since +/// no row will ever point at it. /// /// public sealed partial class FileStorageRelocator( @@ -54,51 +56,63 @@ public sealed partial class FileStorageRelocator( FileAsset? result = null; string? supersededKey = null; + string? copiedKey = null; var strategy = db.Database.CreateExecutionStrategy(); - await strategy.ExecuteAsync(async ct => + try { - // A retried attempt starts from the database again, not from the failed attempt's state. - db.ChangeTracker.Clear(); - result = null; - supersededKey = null; - - await using var transaction = await db.Database.BeginTransactionAsync(ct).ConfigureAwait(false); - var asset = await LoadLockedAsync(fileAssetId, ct).ConfigureAwait(false); - if (asset is null || !mutate(asset)) - { - result = asset; - return; - } - - var oldKey = asset.StorageKey; - var newKey = StorageKeyBuilder.ForVisibility(oldKey, asset.Visibility); - var moves = !string.Equals(oldKey, newKey, StringComparison.Ordinal); - if (moves) + await strategy.ExecuteAsync(async ct => { - // Skip the copy only on a confirmed hit: ExistsAsync reports false on store errors, and the - // copy itself throws on a missing source or a store failure, which aborts before the row moves. - if (!await storage.ExistsAsync(newKey, ct).ConfigureAwait(false)) + // A retried attempt starts from the database again, not from the failed attempt's state. + db.ChangeTracker.Clear(); + result = null; + supersededKey = null; + + await using var transaction = await db.Database.BeginTransactionAsync(ct).ConfigureAwait(false); + var asset = await LoadLockedAsync(fileAssetId, ct).ConfigureAwait(false); + if (asset is null || !mutate(asset)) { - await storage.CopyAsync(oldKey, newKey, ct).ConfigureAwait(false); + result = asset; + return; } - asset.RelocateStorage(newKey); - } + var oldKey = asset.StorageKey; + var newKey = StorageKeyBuilder.ForVisibility(oldKey, asset.Visibility); + var moves = !string.Equals(oldKey, newKey, StringComparison.Ordinal); + if (moves) + { + // Skip the copy only on a confirmed hit: ExistsAsync reports false on store errors, and the + // copy itself throws on a missing source or a store failure, which aborts before the row moves. + if (!await storage.ExistsAsync(newKey, ct).ConfigureAwait(false)) + { + await storage.CopyAsync(oldKey, newKey, ct).ConfigureAwait(false); + } + + copiedKey = newKey; + asset.RelocateStorage(newKey); + } - // acceptAllChangesOnSuccess: false keeps the tracked changes if the commit fails and the - // execution strategy retries; they are accepted once the commit has gone through. - await db.SaveChangesAsync(acceptAllChangesOnSuccess: false, ct).ConfigureAwait(false); - if (moves) - { - await outbox.AddAsync(NewEvent(asset, oldKey, newKey), ct).ConfigureAwait(false); - } + // acceptAllChangesOnSuccess: false keeps the tracked changes if the commit fails and the + // execution strategy retries; they are accepted once the commit has gone through. + await db.SaveChangesAsync(acceptAllChangesOnSuccess: false, ct).ConfigureAwait(false); + if (moves) + { + await outbox.AddAsync(NewEvent(asset, oldKey, newKey), ct).ConfigureAwait(false); + } - await transaction.CommitAsync(ct).ConfigureAwait(false); - db.ChangeTracker.AcceptAllChanges(); + await transaction.CommitAsync(ct).ConfigureAwait(false); + db.ChangeTracker.AcceptAllChanges(); - result = asset; - supersededKey = moves ? oldKey : null; - }, cancellationToken).ConfigureAwait(false); + result = asset; + supersededKey = moves ? oldKey : null; + }, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (copiedKey is not null) + { + // The object was copied but the row never moved to it, so nothing references the copy while + // the quota decorator has charged it. Remove it (RemoveAsync refunds the bytes once it is gone). + await DiscardCopyAsync(fileAssetId, copiedKey, ex).ConfigureAwait(false); + throw; + } if (supersededKey is not null && result is not null) { @@ -157,6 +171,21 @@ await strategy.ExecuteAsync(async ct => }, cancellationToken).ConfigureAwait(false); } + // Best-effort and not tied to the caller's token: the caller's failure is what gets reported. Goes + // through DeleteIfUnreferencedAsync, so a copy a concurrent move has since committed is left alone. + private async Task DiscardCopyAsync(Guid fileAssetId, string copiedKey, Exception failure) + { + LogCommitFailed(logger, failure, fileAssetId, copiedKey); + try + { + await DeleteIfUnreferencedAsync(fileAssetId, copiedKey, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + LogDiscardCopyFailed(logger, ex, fileAssetId, copiedKey); + } + } + // Row lock + fresh read. EF identity resolution would otherwise hand back an already-tracked instance // with stale values, hence the ChangeTracker.Clear() at the start of each attempt. private Task LoadLockedAsync(Guid fileAssetId, CancellationToken ct) @@ -192,6 +221,14 @@ await strategy.ExecuteAsync(async ct => Message = "[Files] moved file {FileAssetId} from {OldStorageKey} to {NewStorageKey}")] private static partial void LogRelocated(ILogger logger, Guid fileAssetId, string oldStorageKey, string newStorageKey); + [LoggerMessage(Level = LogLevel.Warning, + Message = "[Files] move of file {FileAssetId} failed after copying it to {StorageKey}; removing the copy")] + private static partial void LogCommitFailed(ILogger logger, Exception exception, Guid fileAssetId, string storageKey); + + [LoggerMessage(Level = LogLevel.Error, + Message = "[Files] could not remove the unused copy {StorageKey} of file {FileAssetId}; it stays orphaned and charged to quota")] + private static partial void LogDiscardCopyFailed(ILogger logger, Exception exception, Guid fileAssetId, string storageKey); + [LoggerMessage(Level = LogLevel.Warning, Message = "[Files] could not delete the superseded object {StorageKey} of file {FileAssetId} yet; the outbox retries it")] private static partial void LogDeleteDeferred(ILogger logger, Exception exception, Guid fileAssetId, string storageKey); diff --git a/src/Tests/Framework.Tests/Eventing/InMemoryEventBusHandlerIsolationTests.cs b/src/Tests/Framework.Tests/Eventing/InMemoryEventBusHandlerIsolationTests.cs new file mode 100644 index 0000000000..b1ed397b7e --- /dev/null +++ b/src/Tests/Framework.Tests/Eventing/InMemoryEventBusHandlerIsolationTests.cs @@ -0,0 +1,205 @@ +using FSH.Framework.Eventing.Abstractions; +using FSH.Framework.Eventing.Inbox; +using FSH.Framework.Eventing.InMemory; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Framework.Tests.Eventing; + +/// +/// #1423: one failing handler for an event must not stop the other handlers of that event. Every handler +/// runs, the failure still surfaces (so the outbox retries), and the inbox makes the retry re-run only the +/// handler that failed. +/// +public sealed class InMemoryEventBusHandlerIsolationTests +{ + [Fact] + public async Task PublishAsync_Should_RunEveryHandler_And_StillThrow_When_AnEarlierHandlerFails() + { + // Arrange — the failing handler is registered first, as Files' delete runs before Catalog's rewrite. + var failing = new FailingHandler(failuresLeft: 1); + var succeeding = new CountingHandler(); + var inbox = new RecordingInbox(); + using var provider = BuildProvider(inbox, failing, succeeding); + var bus = new InMemoryEventBus(provider, NullLogger.Instance, new NullScope()); + var @event = new TestEvent(); + + // Act + var act = () => bus.PublishAsync(@event); + + // Assert — the later handler ran even though the first one threw, and the failure still surfaced. + (await act.ShouldThrowAsync()).Message.ShouldBe(FailingHandler.Message); + succeeding.Calls.ShouldBe(1); + inbox.Processed.ShouldContain((@event.Id, typeof(CountingHandler).FullName!)); + inbox.Processed.ShouldNotContain((@event.Id, typeof(FailingHandler).FullName!)); + } + + [Fact] + public async Task PublishAsync_Should_ReRunOnlyTheFailedHandler_When_TheEventIsRetried() + { + // Arrange + var failing = new FailingHandler(failuresLeft: 1); + var succeeding = new CountingHandler(); + using var provider = BuildProvider(new RecordingInbox(), failing, succeeding); + var bus = new InMemoryEventBus(provider, NullLogger.Instance, new NullScope()); + var @event = new TestEvent(); + await Should.ThrowAsync(() => bus.PublishAsync(@event)); + + // Act — the outbox's retry publishes the same event again. + await bus.PublishAsync(@event); + + // Assert — the handler that already succeeded is skipped by the inbox; the failed one ran again. + succeeding.Calls.ShouldBe(1); + failing.Calls.ShouldBe(2); + } + + [Fact] + public async Task PublishAsync_Should_ThrowAggregateException_When_SeveralHandlersFail() + { + // Arrange + using var provider = BuildProvider( + new RecordingInbox(), new FailingHandler(failuresLeft: 1), new SecondFailingHandler(), new CountingHandler()); + var bus = new InMemoryEventBus(provider, NullLogger.Instance, new NullScope()); + + // Act + var act = () => bus.PublishAsync(new TestEvent()); + + // Assert + (await act.ShouldThrowAsync()).InnerExceptions.Count.ShouldBe(2); + } + + [Fact] + public async Task PublishAsync_Should_GiveEachHandlerAFreshScope_So_AFailedHandlersPendingChangesAreNotSavedByTheNext() + { + // Arrange - both handlers share a scoped "unit of work", like two handlers on one module DbContext. + // The first stages a change and throws before saving; the second saves whatever is staged. + var saved = new List(); + var services = new ServiceCollection(); + services.AddSingleton(new RecordingInbox()); + services.AddSingleton(saved); + services.AddScoped(); + services.AddScoped, StagesThenThrowsHandler>(); + services.AddScoped, SavesStagedHandler>(); + using var provider = services.BuildServiceProvider(); + var bus = new InMemoryEventBus(provider, NullLogger.Instance, new NullScope()); + + // Act + await Should.ThrowAsync(() => bus.PublishAsync(new TestEvent())); + + // Assert - the second handler ran in its own scope, so the failed handler's staged change was dropped. + saved.ShouldBe(["saved-by-second"]); + } + + private static ServiceProvider BuildProvider(IInboxStore inbox, params IIntegrationEventHandler[] handlers) + { + var services = new ServiceCollection(); + services.AddSingleton(inbox); + foreach (var handler in handlers) + { + services.AddSingleton(handler); + } + + return services.BuildServiceProvider(); + } + + #region Test doubles + + private sealed record TestEvent : IIntegrationEvent + { + public Guid Id { get; } = Guid.CreateVersion7(); + public DateTime OccurredOnUtc { get; } = DateTime.UtcNow; + public string? TenantId { get; } = "acme"; + public string CorrelationId { get; } = Guid.CreateVersion7().ToString(); + public string Source { get; } = "tests"; + } + + private sealed class FailingHandler(int failuresLeft) : IIntegrationEventHandler + { + public const string Message = "store rejected the delete"; + private int _failuresLeft = failuresLeft; + + public int Calls { get; private set; } + + public async Task HandleAsync(TestEvent @event, CancellationToken ct = default) + { + await Task.Yield(); + Calls++; + if (_failuresLeft-- > 0) + { + throw new InvalidOperationException(Message); + } + } + } + + private sealed class SecondFailingHandler : IIntegrationEventHandler + { + public async Task HandleAsync(TestEvent @event, CancellationToken ct = default) + { + await Task.Yield(); + throw new InvalidOperationException("second handler failed"); + } + } + + private sealed class CountingHandler : IIntegrationEventHandler + { + public int Calls { get; private set; } + + public Task HandleAsync(TestEvent @event, CancellationToken ct = default) + { + Calls++; + return Task.CompletedTask; + } + } + + private sealed class UnitOfWork + { + public List Staged { get; } = []; + } + + private sealed class StagesThenThrowsHandler(UnitOfWork work) : IIntegrationEventHandler + { + public Task HandleAsync(TestEvent @event, CancellationToken ct = default) + { + work.Staged.Add("partial-from-failed-handler"); + throw new InvalidOperationException("failed before saving"); + } + } + + private sealed class SavesStagedHandler(UnitOfWork work, List saved) : IIntegrationEventHandler + { + public Task HandleAsync(TestEvent @event, CancellationToken ct = default) + { + work.Staged.Add("saved-by-second"); + saved.AddRange(work.Staged); + return Task.CompletedTask; + } + } + + private sealed class RecordingInbox : IInboxStore + { + public HashSet<(Guid EventId, string Handler)> Processed { get; } = []; + + public Task HasProcessedAsync(Guid eventId, string handlerName, CancellationToken ct = default) + => Task.FromResult(Processed.Contains((eventId, handlerName))); + + public Task MarkProcessedAsync(Guid eventId, string handlerName, string? tenantId, string eventType, CancellationToken ct = default) + { + Processed.Add((eventId, handlerName)); + return Task.CompletedTask; + } + } + + private sealed class NullScope : IEventTenantScope + { + public IDisposable Begin(string? tenantId) => new Noop(); + + private sealed class Noop : IDisposable + { + public void Dispose() + { + } + } + } + + #endregion +} diff --git a/src/Tests/Framework.Tests/Storage/S3PrefixValidationTests.cs b/src/Tests/Framework.Tests/Storage/S3PrefixValidationTests.cs new file mode 100644 index 0000000000..1f544d4c9d --- /dev/null +++ b/src/Tests/Framework.Tests/Storage/S3PrefixValidationTests.cs @@ -0,0 +1,95 @@ +using FSH.Framework.Storage; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Framework.Tests.Storage; + +/// +/// #1423: keys are {Prefix}/public/… and {Prefix}/private/…. A prefix whose first segment is +/// itself public puts every private key under public/, where the anonymous-read policy applies, +/// so such a prefix has to fail at startup rather than quietly expose private files. +/// +public sealed class S3PrefixValidationTests +{ + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData("tenant-files")] + [InlineData("apps/fsh")] + [InlineData("publicity")] + [InlineData("files/public")] + public void AddHeroStorage_Should_PassStartupValidation_When_PrefixDoesNotStartWithAVisibilityRoot(string? prefix) + { + // Arrange + using var provider = BuildProvider(prefix); + + // Act + provider.GetRequiredService().Validate(); + + // Assert + provider.GetRequiredService>().Value.Prefix.ShouldBe(prefix); + } + + [Theory] + [InlineData("public")] + [InlineData("private")] + [InlineData("Public")] + [InlineData("PRIVATE/")] + [InlineData("public/apps")] + [InlineData("/private/apps")] + [InlineData(" public ")] + public void AddHeroStorage_Should_FailAtStartup_When_PrefixStartsWithAVisibilityRoot(string prefix) + { + // Arrange + using var provider = BuildProvider(prefix); + + // Act + var act = () => provider.GetRequiredService().Validate(); + + // Assert + act.ShouldThrow() + .Failures.ShouldContain(failure => failure.Contains("Storage:S3:Prefix", StringComparison.Ordinal)); + } + + // The rule belongs to the S3 provider; a host on local storage (the DbMigrator, dev) never sees it. + [Fact] + public void AddHeroStorage_Should_NotValidatePrefix_When_ProviderIsLocal() + { + // Arrange + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["Storage:Provider"] = "local", + ["Storage:S3:Prefix"] = "public", + }).Build(); + var services = new ServiceCollection(); + services.AddLogging(); + services.AddHeroStorage(configuration); + using var provider = services.BuildServiceProvider(); + + // Act — local storage registers no S3 options validation, so there may be no validator at all. + var act = () => provider.GetService()?.Validate(); + + // Assert + act.ShouldNotThrow(); + } + + private static ServiceProvider BuildProvider(string? prefix) + { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["Storage:Provider"] = "s3", + ["Storage:S3:Bucket"] = "prefix-tests", + ["Storage:S3:Region"] = "us-east-1", + ["Storage:S3:ServiceUrl"] = "http://internal-store:9000", + ["Storage:S3:AccessKey"] = "test-access-key", + ["Storage:S3:SecretKey"] = "test-secret-key", + ["Storage:S3:Prefix"] = prefix, + }).Build(); + + var services = new ServiceCollection(); + services.AddLogging(); + services.AddHeroStorage(configuration); + return services.BuildServiceProvider(); + } +} diff --git a/src/Tests/Integration.Tests/Tests/Files/StorageVisibilityRootTests.cs b/src/Tests/Integration.Tests/Tests/Files/StorageVisibilityRootTests.cs index 098d00a80f..5731f81b26 100644 --- a/src/Tests/Integration.Tests/Tests/Files/StorageVisibilityRootTests.cs +++ b/src/Tests/Integration.Tests/Tests/Files/StorageVisibilityRootTests.cs @@ -4,18 +4,25 @@ using Finbuckle.MultiTenant; using Finbuckle.MultiTenant.Abstractions; using FSH.Framework.Core.Domain; +using FSH.Framework.Quota; using FSH.Framework.Shared.Multitenancy; +using FSH.Framework.Shared.Quota; +using FSH.Framework.Storage; using FSH.Framework.Storage.Services; using FSH.Modules.Catalog.Data; using FSH.Modules.Catalog.Domain; +using FSH.Modules.Files.Contracts.v1.Commands; using FSH.Modules.Files.Contracts.v1.DTOs; using FSH.Modules.Files.Data; using FSH.Modules.Files.Domain; +using FSH.Modules.Files.Features.v1.FinalizeUpload; using FSH.Modules.Files.Jobs; +using FSH.Modules.Files.Services; using FSH.Modules.Identity.Data; using FSH.Modules.Multitenancy.Contracts.Dtos; using Integration.Tests.Infrastructure; using Integration.Tests.Infrastructure.Extensions; +using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; namespace Integration.Tests.Tests.Files; @@ -365,10 +372,161 @@ await s3.PutObjectAsync(new PutObjectRequest (await ObjectExistsAsync(legacyKey)).ShouldBeFalse(); } + [Fact] + public async Task Finalize_Should_Succeed_And_LeaveTheFileAtItsLegacyKey_When_TheMoveAfterCommitFails() + { + // Arrange — an in-flight public upload at a pre-#1410 key, bytes already PUT. + var adminId = await GetUserIdAsync(TestConstants.RootTenantId, TestConstants.RootAdminEmail); + var (id, legacyKey, bytes) = await SeedPendingLegacyUploadAsync(adminId, "move-fails.pdf"); + + // Act — finalize through a store whose copy fails, so the move after the finalize commit throws. + FileAssetDto dto; + using (var scope = _factory.Services.CreateScope()) + { + SetTenant(scope, TestConstants.RootTenantId); + var failing = new CopyFailsStorage(scope.ServiceProvider.GetRequiredService()); + var relocator = ActivatorUtilities.CreateInstance(scope.ServiceProvider, failing); + var handler = ActivatorUtilities.CreateInstance( + scope.ServiceProvider, failing, relocator, new FixedCurrentUser(TestConstants.RootTenantId, Guid.Parse(adminId))); + dto = await handler.Handle(new FinalizeUploadCommand(id), CancellationToken.None); + failing.FailedCopies.ShouldBe(1); + } + + // Assert — the client gets the committed, Available file instead of a 500, still at its legacy key. + dto.Status.ShouldBe(FileAssetStatus.Available); + (await ReadStorageKeyAsync(TestConstants.RootTenantId, id)).ShouldBe(legacyKey); + (await ReadObjectAsync(legacyKey)).ShouldBe(bytes); + (await ObjectExistsAsync("public/" + legacyKey)).ShouldBeFalse(); + + // …and the legacy-key migration repairs it on its next run. + await RunMigrationJobAsync(); + await OutboxDrain.DrainAsync(_factory.Services); + (await ReadStorageKeyAsync(TestConstants.RootTenantId, id)).ShouldBe("public/" + legacyKey); + (await ReadObjectAsync("public/" + legacyKey)).ShouldBe(bytes); + } + + [Fact] + public async Task ChangeVisibility_Should_RemoveTheCopy_And_RefundItsBytes_When_TheCommitFailsAfterTheCopy() + { + // Arrange — a private upload, and a quota-metered store so the copy is charged as in production. + using var client = await _auth.CreateRootAdminClientAsync(); + var bytes = RandomBytes(512); + var id = await UploadAndFinalizeAsync(client, "commit-fails.pdf", bytes, visibility: 1); + var privateKey = await ReadStorageKeyAsync(TestConstants.RootTenantId, id); + var publicKey = "public/" + privateKey["private/".Length..]; + var quotaOptions = new QuotaOptions { Enabled = true }; + var quotas = new InMemoryQuotaService( + new InMemoryQuotaStore(), quotaOptions, new QuotaPlanResolver(quotaOptions), [], TimeProvider.System); + + // Act — make it public; the copy succeeds, then the transaction fails before it commits. + var outbox = new ThrowingOutboxWriter(); + using (var scope = _factory.Services.CreateScope()) + { + SetTenant(scope, TestConstants.RootTenantId); + var metered = new QuotaMeteredStorageService( + scope.ServiceProvider.GetRequiredService(), + quotas, + scope.ServiceProvider.GetRequiredService>(), + NullLogger.Instance); + var relocator = ActivatorUtilities.CreateInstance(scope.ServiceProvider, metered, outbox); + + await Should.ThrowAsync(() => relocator.ApplyAsync(id, asset => + { + asset.ChangeVisibility(Visibility.Public); + return true; + })); + } + + // Assert — the copy was made (the outbox write comes after it), then removed and refunded. + outbox.Calls.ShouldBe(1); + (await ReadStorageKeyAsync(TestConstants.RootTenantId, id)).ShouldBe(privateKey); + (await ReadVisibilityAsync(TestConstants.RootTenantId, id)).ShouldBe(Visibility.Private); + (await ReadObjectAsync(privateKey)).ShouldBe(bytes); + (await ObjectExistsAsync(publicKey)).ShouldBeFalse("a copy no row points at must not be left behind"); + (await quotas.GetCurrentAsync(TestConstants.RootTenantId, QuotaResource.StorageBytes)) + .ShouldBe(0, "the copy's charge is refunded when the copy is removed"); + } + // ─── helpers ───────────────────────────────────────────────────── private sealed record LegacyAsset(Guid Id, string Key, byte[] Bytes); + private async Task<(Guid Id, string Key, byte[] Bytes)> SeedPendingLegacyUploadAsync(string adminId, string fileName) + { + var id = Guid.CreateVersion7(); + var legacyKey = $"tenants/{TestConstants.RootTenantId}/myfiles/2025/01/{id:N}/{fileName}"; + var bytes = RandomBytes(300); + + var s3 = _factory.Services.GetRequiredService(); + using (var stream = new MemoryStream(bytes)) + { + await s3.PutObjectAsync(new PutObjectRequest + { + BucketName = Bucket, + Key = legacyKey, + InputStream = stream, + ContentType = "application/pdf", + }); + } + + using var scope = _factory.Services.CreateScope(); + SetTenant(scope, TestConstants.RootTenantId); + var db = scope.ServiceProvider.GetRequiredService(); + db.FileAssets.Add(FileAsset.CreatePending( + id, "MyFiles", null, fileName, fileName, "application/pdf", bytes.Length, legacyKey, + Visibility.Public, createdByUserId: adminId, uploadDeadline: DateTimeOffset.UtcNow.AddMinutes(15))); + await db.SaveChangesAsync(); + return (id, legacyKey, bytes); + } + + /// A store whose server-side copy fails, as when the store rejects the request. + private sealed class CopyFailsStorage(IStorageService inner) : IStorageService + { + public int FailedCopies { get; private set; } + + public Task CopyAsync(string sourceKey, string destinationKey, CancellationToken cancellationToken = default) + { + FailedCopies++; + throw new AmazonS3Exception("copy rejected by the store"); + } + + public Task RemoveAsync(string path, CancellationToken cancellationToken = default) => inner.RemoveAsync(path, cancellationToken); + public Task UploadAsync(FSH.Framework.Shared.Storage.FileUploadRequest request, FSH.Framework.Storage.FileType fileType, CancellationToken cancellationToken = default) + where T : class => inner.UploadAsync(request, fileType, cancellationToken); + public Task DownloadAsync(string path, CancellationToken cancellationToken = default) => inner.DownloadAsync(path, cancellationToken); + public Task ExistsAsync(string path, CancellationToken cancellationToken = default) => inner.ExistsAsync(path, cancellationToken); + public Task GetSizeAsync(string path, CancellationToken cancellationToken = default) => inner.GetSizeAsync(path, cancellationToken); + public Task GenerateUploadUrlAsync(string storageKey, string contentType, long maxBytes, TimeSpan ttl, CancellationToken cancellationToken = default) + => inner.GenerateUploadUrlAsync(storageKey, contentType, maxBytes, ttl, cancellationToken); + public Task GenerateDownloadUrlAsync(string storageKey, TimeSpan ttl, string? responseContentDisposition = null, CancellationToken cancellationToken = default) + => inner.GenerateDownloadUrlAsync(storageKey, ttl, responseContentDisposition, cancellationToken); + public Task HeadObjectAsync(string storageKey, CancellationToken cancellationToken = default) => inner.HeadObjectAsync(storageKey, cancellationToken); + public string BuildPublicUrl(string storageKey) => inner.BuildPublicUrl(storageKey); + } + + /// An outbox write that fails, which aborts the move's transaction before it commits. + private sealed class ThrowingOutboxWriter : FSH.Framework.Eventing.Abstractions.IOutboxWriter + { + public int Calls { get; private set; } + + public Task AddAsync(FSH.Framework.Eventing.Abstractions.IIntegrationEvent @event, CancellationToken ct = default) + { + Calls++; + throw new InvalidOperationException("the transaction failed before commit"); + } + } + + private sealed class FixedCurrentUser(string tenantId, Guid userId) : FSH.Framework.Core.Context.ICurrentUser + { + public string? Name => null; + public Guid GetUserId() => userId; + public string? GetUserEmail() => null; + public string? GetTenant() => tenantId; + public bool IsAuthenticated() => true; + public bool IsInRole(string role) => false; + public IEnumerable? GetUserClaims() => null; + } + private static byte[] RandomBytes(int size) { byte[] bytes = new byte[size];