Convert EditFailedMessageManager to an atomic write operation

docs/coding-and-design-guidelines.md

@@ -40,6 +40,23 @@ There are a few things that are still registered using convention. Note that the
 Additionally, because NServiceBus does a type-scan at startup it will automatically register any implementations of `Feature` and `IHandleMessage<>`. We have chosen to leave this alone as we would be fighting with NServiceBus in order to turn this off. 
 
 
+## Prefer explicit persistence operations
+
+Use a direct data-store method when all inputs for a persistence operation fit in one method call. The method should own and dispose its EF Core scope/context or RavenDB session, accept a `CancellationToken`, and commit before returning. Returned entities are detached snapshots; callers must not be required to mutate tracked entities as an implicit persistence command.
+
+Atomic operations that can encounter concurrency conflicts should document their provider guarantees and translate expected provider exceptions into explicit domain outcomes. For example, a unique-key or optimistic-concurrency conflict should not escape when contention is part of the operation's normal contract.
+
+Reserve a specialized unit of work for cases where a caller genuinely composes several writes into one atomic batch. New unit-of-work APIs should consistently provide:
+
+- an `I...UnitOfWorkFactory`;
+- a `StartNew` factory method;
+- a `Complete(CancellationToken)` commit method;
+- `IAsyncDisposable` lifetime ownership;
+- explicit operation-recording methods rather than mutation of tracked return values;
+- documented commit, abandon, repeated-completion, and concurrency behavior.
+
+Do not introduce generic `IDataSessionManager`-style abstractions or persistence managers with hidden call-order protocols. During review, prefer one explicit store operation unless caller-composed atomicity requires a unit of work.
+
 ## Avoid property injection
 
 Although the Autofac container can be configured to allow property injection, we prefer to avoid it. There is no way to specify property injection using the Microsoft DI abstractions, and the default `IServiceProvider` implementation does not support it. Where possible, use constructor injection instead.

src/ServiceControl.Persistence.EFCore/Implementation/EditFailedMessagesDataStore.cs

@@ -1,14 +1,99 @@
 namespace ServiceControl.Persistence.EFCore.Implementation;
 
-using DbContexts;
+using Entities;
+using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
+using ServiceControl.MessageFailures;
 
-public class EditFailedMessagesDataStore(IServiceScopeFactory scopeFactory, TimeProvider timeProvider) : IEditFailedMessagesDataStore
+public class EditFailedMessagesDataStore(IServiceScopeFactory scopeFactory, TimeProvider timeProvider)
+    : DataStoreBase(scopeFactory), IEditFailedMessagesDataStore
 {
-    public Task<IEditFailedMessagesManager> CreateEditFailedMessageManager(CancellationToken cancellationToken = default)
+    public Task<string?> GetCurrentEditingRequestId(string failedMessageId, CancellationToken cancellationToken = default)
     {
-        var scope = scopeFactory.CreateAsyncScope();
-        return Task.FromResult<IEditFailedMessagesManager>(
-            new EditFailedMessagesManager(scope, scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>(), timeProvider));
+        if (!Guid.TryParse(failedMessageId, out var uniqueMessageId))
+        {
+            return Task.FromResult<string?>(null);
+        }
+
+        return ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessageEdits
+            .AsNoTracking()
+            .Where(edit => edit.UniqueMessageId == uniqueMessageId)
+            .Select(edit => edit.EditId)
+            .SingleOrDefaultAsync(token), cancellationToken);
+    }
+
+    public Task<BeginEditResult> TryBeginEdit(string failedMessageId, string editingMessageId, CancellationToken cancellationToken = default)
+    {
+        if (!Guid.TryParse(failedMessageId, out var uniqueMessageId))
+        {
+            return Task.FromResult(new BeginEditResult(BeginEditOutcome.MessageNotFound));
+        }
+
+        return ExecuteWithDbContext(async (dbContext, ct) =>
+            {
+                var entity = await dbContext.FailedMessages
+                    .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, ct);
+
+                if (entity is null)
+                {
+                    return new BeginEditResult(BeginEditOutcome.MessageNotFound);
+                }
+
+                var existingEditId = await dbContext.FailedMessageEdits
+                    .Where(edit => edit.UniqueMessageId == uniqueMessageId)
+                    .Select(edit => edit.EditId)
+                    .SingleOrDefaultAsync(ct);
+
+                if (existingEditId is not null)
+                {
+                    return existingEditId == editingMessageId
+                        ? new BeginEditResult(BeginEditOutcome.Acquired, entity.ToFailedMessage([]), existingEditId)
+                        : new BeginEditResult(BeginEditOutcome.AcquiredByAnotherEdit, ExistingEditId: existingEditId);
+                }
+
+                if (entity.Status != FailedMessageStatus.Unresolved)
+                {
+                    return new BeginEditResult(BeginEditOutcome.MessageNotUnresolved);
+                }
+
+                dbContext.FailedMessageEdits.Add(new FailedMessageEditEntity { UniqueMessageId = uniqueMessageId, EditId = editingMessageId });
+
+                var now = timeProvider.GetUtcNow().UtcDateTime;
+                entity.Status = FailedMessageStatus.Resolved;
+                entity.StatusChangedAt = now;
+                entity.LastModified = now;
+
+                try
+                {
+                    await dbContext.SaveChangesAsync(ct);
+                    return new BeginEditResult(BeginEditOutcome.Acquired, entity.ToFailedMessage([]));
+                }
+                catch (DbUpdateException exception) when (dbContext.IsDuplicateKeyException(exception))
+                {
+                    dbContext.ChangeTracker.Clear();
+
+                    var winningEditId = await dbContext.FailedMessageEdits
+                        .AsNoTracking()
+                        .Where(edit => edit.UniqueMessageId == uniqueMessageId)
+                        .Select(edit => edit.EditId)
+                        .SingleOrDefaultAsync(ct);
+
+                    if (winningEditId is null)
+                    {
+                        throw;
+                    }
+
+                    if (winningEditId != editingMessageId)
+                    {
+                        return new BeginEditResult(BeginEditOutcome.AcquiredByAnotherEdit, ExistingEditId: winningEditId);
+                    }
+
+                    entity = await dbContext.FailedMessages
+                        .AsNoTracking()
+                        .SingleAsync(message => message.UniqueMessageId == uniqueMessageId, ct);
+
+                    return new BeginEditResult(BeginEditOutcome.Acquired, entity.ToFailedMessage([]), winningEditId);
+                }
+            }, cancellationToken);
     }
 }

src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs

@@ -2,11 +2,70 @@ namespace ServiceControl.Persistence.RavenDB.Editing
 {
     using System.Threading;
     using System.Threading.Tasks;
+    using Raven.Client.Exceptions;
+    using ServiceControl.MessageFailures;
+    using ServiceControl.Persistence.Recoverability.Editing;
 
     class EditFailedMessagesDataStore(IRavenSessionProvider sessionProvider, ExpirationManager expirationManager) : IEditFailedMessagesDataStore
     {
-        public async Task<IEditFailedMessagesManager> CreateEditFailedMessageManager(CancellationToken cancellationToken = default) =>
-            // the edit failed message manager manages the lifetime of the session
-            new EditFailedMessageManager(await sessionProvider.OpenSession(cancellationToken: cancellationToken), expirationManager);
+        public async Task<string> GetCurrentEditingRequestId(string failedMessageId, CancellationToken cancellationToken = default)
+        {
+            using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
+            var edit = await session.LoadAsync<FailedMessageEdit>(FailedMessageEdit.MakeDocumentId(failedMessageId), cancellationToken);
+            return edit?.EditId;
+        }
+
+        public async Task<BeginEditResult> TryBeginEdit(string failedMessageId, string editingMessageId, CancellationToken cancellationToken = default)
+        {
+            using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
+            session.Advanced.UseOptimisticConcurrency = true;
+
+            var failedMessage = await session.LoadAsync<FailedMessage>(FailedMessageIdGenerator.MakeDocumentId(failedMessageId), cancellationToken);
+            if (failedMessage is null)
+            {
+                return new BeginEditResult(BeginEditOutcome.MessageNotFound);
+            }
+            var editDocumentId = FailedMessageEdit.MakeDocumentId(failedMessageId);
+            var existingEdit = await session.LoadAsync<FailedMessageEdit>(editDocumentId, cancellationToken);
+            if (existingEdit is not null)
+            {
+                return existingEdit.EditId == editingMessageId
+                    ? new BeginEditResult(BeginEditOutcome.Acquired, failedMessage, existingEdit.EditId)
+                    : new BeginEditResult(BeginEditOutcome.AcquiredByAnotherEdit, ExistingEditId: existingEdit.EditId);
+            }
+
+            if (failedMessage.Status != FailedMessageStatus.Unresolved)
+            {
+                return new BeginEditResult(BeginEditOutcome.MessageNotUnresolved);
+            }
+
+            await session.StoreAsync(new FailedMessageEdit { Id = editDocumentId, FailedMessageId = failedMessage.Id, EditId = editingMessageId }, cancellationToken);
+
+            failedMessage.Status = FailedMessageStatus.Resolved;
+            expirationManager.EnableExpiration(session, failedMessage);
+
+            try
+            {
+                await session.SaveChangesAsync(cancellationToken);
+                return new(BeginEditOutcome.Acquired, failedMessage);
+            }
+            catch (ConcurrencyException)
+            {
+                // One bounded reload is sufficient: Raven reports the conflict only after the
+                // competing atomic batch has won, so the persisted claim identifies the outcome.
+                existingEdit = await ReloadConflictResult(failedMessageId, cancellationToken);
+            }
+
+            return existingEdit.EditId == editingMessageId
+                ? new BeginEditResult(BeginEditOutcome.Acquired, failedMessage, existingEdit.EditId)
+                : new BeginEditResult(BeginEditOutcome.AcquiredByAnotherEdit, ExistingEditId: existingEdit.EditId);
+        }
+
+        async Task<FailedMessageEdit> ReloadConflictResult(string failedMessageId, CancellationToken cancellationToken)
+        {
+            using var reloadSession = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
+            return await reloadSession.LoadAsync<FailedMessageEdit>(FailedMessageEdit.MakeDocumentId(failedMessageId), cancellationToken);
+        }
+
     }
 }