fix(files): isolate event handlers, harden storage moves, validate S3 prefix

.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):

.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<T>` 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<T>` 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<T>` 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`.**

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<ExceptionDispatchInfo>? 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<IInboxStore>();
+                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<IInboxStore>();
+            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<ExceptionDispatchInfo>? 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)

src/BuildingBlocks/Storage/Extensions.cs

@@ -38,6 +38,7 @@ public static IServiceCollection AddHeroStorage(this IServiceCollection services
                 .AddOptions<S3StorageOptions>()
                 .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<IAmazonS3>(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<TInner>(
         IServiceCollection services,
         bool quotaEnabled,

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<FinalizeUploadCommandHandler> logger)
     : ICommandHandler<FinalizeUploadCommand, FileAssetDto>
 {
     public async ValueTask<FileAssetDto> 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<FileAsset> 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);
 }