Update from reviewer comments:

* Make outbox truly generic, handling multiple queues * Add "MoveAsync" to SharedStorage * Refactor saving pretranslations file
sillsdev · Jun 27, 2024 · be36554 · be36554
1 parent 1d5e603
commit be36554
Show file tree

Hide file tree

Showing 21 changed files with 352 additions and 240 deletions.
diff --git a/src/SIL.Machine.AspNetCore/Configuration/IMachineBuilderExtensions.cs b/src/SIL.Machine.AspNetCore/Configuration/IMachineBuilderExtensions.cs
@@ -279,7 +279,7 @@ await c.Indexes.CreateOrUpdateAsync(
                         )
                 );
                 o.AddRepository<OutboxMessage>("outbox_messages");
-                o.AddRepository<SortableIndex>("sortable_indexes");
+                o.AddRepository<Outbox>("outboxes");
             }
         );
         builder.Services.AddHealthChecks().AddMongoDb(connectionString!, name: "Mongo");
@@ -298,6 +298,8 @@ public static IMachineBuilder AddServalPlatformService(
 
         builder.Services.AddScoped<IPlatformService, ServalPlatformService>();
 
+        builder.Services.AddSingleton<IOutboxMessageHandler, ServalPlatformOutboxHandler>();
+
         builder.Services.AddSingleton<IMessageOutboxService, MessageOutboxService>();
 
         builder

diff --git a/src/SIL.Machine.AspNetCore/Models/Outbox.cs b/src/SIL.Machine.AspNetCore/Models/Outbox.cs
@@ -0,0 +1,28 @@
+namespace SIL.Machine.AspNetCore.Models;
+
+public record Outbox : IEntity
+{
+    public string Id { get; set; } = "";
+
+    public int Revision { get; set; }
+
+    public string Name { get; set; } = null!;
+    public int CurrentIndex { get; set; }
+
+    public static async Task<Outbox> GetOutboxNextIndexAsync(
+        IRepository<Outbox> indexRepository,
+        string outboxName,
+        CancellationToken cancellationToken
+    )
+    {
+        Outbox outbox = (
+            await indexRepository.UpdateAsync(
+                i => i.Name == outboxName,
+                i => i.Inc(b => b.CurrentIndex, 1),
+                upsert: true,
+                cancellationToken: cancellationToken
+            )
+        )!;
+        return outbox;
+    }
+}
diff --git a/src/SIL.Machine.AspNetCore/Models/OutboxMessage.cs b/src/SIL.Machine.AspNetCore/Models/OutboxMessage.cs
@@ -1,22 +1,12 @@
 namespace SIL.Machine.AspNetCore.Models;
 
-public enum OutboxMessageMethod
-{
-    BuildStarted,
-    BuildCompleted,
-    BuildCanceled,
-    BuildFaulted,
-    BuildRestarting,
-    InsertPretranslations,
-    IncrementTranslationEngineCorpusSize
-}
-
 public record OutboxMessage : IEntity
 {
     public string Id { get; set; } = "";
     public int Revision { get; set; } = 1;
-    public required string SortableIndex { get; set; }
-    public required OutboxMessageMethod Method { get; set; }
+    public required int Index { get; set; }
+    public required string OutboxName { get; set; }
+    public required string Method { get; set; }
     public required string GroupId { get; set; }
     public required string? RequestContent { get; set; }
     public DateTimeOffset Created { get; set; } = DateTimeOffset.UtcNow;

diff --git a/src/SIL.Machine.AspNetCore/Models/SortableIndex.cs b/src/SIL.Machine.AspNetCore/Models/SortableIndex.cs
diff --git a/src/SIL.Machine.AspNetCore/Services/IFileStorage.cs b/src/SIL.Machine.AspNetCore/Services/IFileStorage.cs
@@ -16,5 +16,6 @@ Task<IReadOnlyCollection<string>> ListFilesAsync(
 
     Task<string> GetDownloadUrlAsync(string path, DateTime expiresAt, CancellationToken cancellationToken = default);
 
+    Task MoveAsync(string sourcePath, string destPath, CancellationToken cancellationToken = default);
     Task DeleteAsync(string path, bool recurse = false, CancellationToken cancellationToken = default);
 }
diff --git a/src/SIL.Machine.AspNetCore/Services/IMessageOutboxService.cs b/src/SIL.Machine.AspNetCore/Services/IMessageOutboxService.cs
@@ -2,11 +2,11 @@
 
 public interface IMessageOutboxService
 {
-    public Task<string> EnqueueMessageAsync(
-        OutboxMessageMethod method,
+    public Task<string> EnqueueMessageAsync<T>(
+        T method,
         string groupId,
-        string requestContent,
-        bool alwaysSaveContentToDisk = false,
+        string? requestContent = null,
+        string? requestContentPath = null,
         CancellationToken cancellationToken = default
     );
 }
diff --git a/src/SIL.Machine.AspNetCore/Services/IOutboxMessageHandler.cs b/src/SIL.Machine.AspNetCore/Services/IOutboxMessageHandler.cs
@@ -0,0 +1,9 @@
+namespace SIL.Machine.AspNetCore.Services;
+
+public interface IOutboxMessageHandler
+{
+    public string Name { get; }
+
+    public Task SendMessageAsync(OutboxMessage message, CancellationToken cancellationToken = default);
+    public Task CleanupMessageAsync(OutboxMessage message, CancellationToken cancellationToken = default);
+}
diff --git a/src/SIL.Machine.AspNetCore/Services/IPlatformService.cs b/src/SIL.Machine.AspNetCore/Services/IPlatformService.cs
@@ -22,9 +22,5 @@ Task BuildCompletedAsync(
     Task BuildFaultedAsync(string buildId, string message, CancellationToken cancellationToken = default);
     Task BuildRestartingAsync(string buildId, CancellationToken cancellationToken = default);
 
-    Task InsertPretranslationsAsync(
-        string engineId,
-        IAsyncEnumerable<Pretranslation> pretranslations,
-        CancellationToken cancellationToken = default
-    );
+    Task InsertPretranslationsAsync(string engineId, string path, CancellationToken cancellationToken = default);
 }
diff --git a/src/SIL.Machine.AspNetCore/Services/ISharedFileService.cs b/src/SIL.Machine.AspNetCore/Services/ISharedFileService.cs
@@ -19,6 +19,7 @@ Task<IReadOnlyCollection<string>> ListFilesAsync(
     Task<Stream> OpenWriteAsync(string path, CancellationToken cancellationToken = default);
 
     Task<bool> ExistsAsync(string path, CancellationToken cancellationToken = default);
+    Task MoveAsync(string sourcePath, string destPath, CancellationToken cancellationToken = default);
 
     Task DeleteAsync(string path, CancellationToken cancellationToken = default);
 }
diff --git a/src/SIL.Machine.AspNetCore/Services/InMemoryStorage.cs b/src/SIL.Machine.AspNetCore/Services/InMemoryStorage.cs
@@ -133,6 +133,15 @@ public async Task DeleteAsync(string path, bool recurse, CancellationToken cance
         }
     }
 
+    public Task MoveAsync(string sourcePath, string destPath, CancellationToken cancellationToken = default)
+    {
+        if (!_memoryStreams.TryGetValue(Normalize(sourcePath), out Entry? entry))
+            throw new FileNotFoundException($"Unable to find file {sourcePath}");
+        _memoryStreams[Normalize(destPath)] = entry;
+        _memoryStreams.Remove(Normalize(sourcePath), out _);
+        return Task.CompletedTask;
+    }
+
     protected override void DisposeManagedResources()
     {
         foreach (Entry stream in _memoryStreams.Values)

diff --git a/src/SIL.Machine.AspNetCore/Services/LocalStorage.cs b/src/SIL.Machine.AspNetCore/Services/LocalStorage.cs
@@ -59,6 +59,15 @@ public Task<Stream> OpenWriteAsync(string path, CancellationToken cancellationTo
         return Task.FromResult<Stream>(File.OpenWrite(pathUri.LocalPath));
     }
 
+    public Task MoveAsync(string sourcePath, string destPath, CancellationToken cancellationToken = default)
+    {
+        Uri sourcePathUri = new(_basePath, Normalize(sourcePath));
+        Uri destPathUri = new(_basePath, Normalize(destPath));
+        Directory.CreateDirectory(Path.GetDirectoryName(destPathUri.LocalPath)!);
+        File.Move(sourcePathUri.LocalPath, destPathUri.LocalPath);
+        return Task.CompletedTask;
+    }
+
     public async Task DeleteAsync(string path, bool recurse, CancellationToken cancellationToken = default)
     {
         Uri pathUri = new(_basePath, Normalize(path));

diff --git a/src/SIL.Machine.AspNetCore/Services/MessageOutboxDeliveryService.cs b/src/SIL.Machine.AspNetCore/Services/MessageOutboxDeliveryService.cs
@@ -1,18 +1,16 @@
-using Serval.Translation.V1;
-
-namespace SIL.Machine.AspNetCore.Services;
+namespace SIL.Machine.AspNetCore.Services;
 
 public class MessageOutboxDeliveryService(
-    TranslationPlatformApi.TranslationPlatformApiClient client,
     IRepository<OutboxMessage> messages,
-    ISharedFileService sharedFileService,
+    IEnumerable<IOutboxMessageHandler> outboxMessageHandlers,
     MessageOutboxOptions options,
     ILogger<MessageOutboxDeliveryService> logger
 ) : BackgroundService
 {
-    private readonly TranslationPlatformApi.TranslationPlatformApiClient _client = client;
     private readonly IRepository<OutboxMessage> _messages = messages;
-    private readonly ISharedFileService _sharedFileService = sharedFileService;
+    private readonly Dictionary<string, IOutboxMessageHandler> _outboxMessageHandlers =
+        outboxMessageHandlers.ToDictionary(o => o.Name);
+
     private readonly ILogger<MessageOutboxDeliveryService> _logger = logger;
     protected TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(10);
     protected TimeSpan MessageExpiration { get; set; } = TimeSpan.FromHours(options.MessageExpirationInHours);
@@ -38,19 +36,20 @@ protected async Task ProcessMessagesAsync(CancellationToken cancellationToken =
         IReadOnlyList<OutboxMessage> messages = await _messages.GetAllAsync();
 
         IEnumerable<List<OutboxMessage>> messageGroups = messages.GroupBy(
-            m => m.GroupId,
+            m => new { m.GroupId, m.OutboxName },
             m => m,
-            (key, element) => element.OrderBy(m => m.SortableIndex).ToList()
+            (key, element) => element.OrderBy(m => m.Index).ToList()
         );
 
         foreach (List<OutboxMessage> messageGroup in messageGroups)
         {
             bool abortMessageGroup = false;
+            var outboxMessageHandler = _outboxMessageHandlers[messageGroup.First().OutboxName];
             foreach (OutboxMessage message in messageGroup)
             {
                 try
                 {
-                    await ProcessGroupMessagesAsync(message, cancellationToken);
+                    await ProcessGroupMessagesAsync(message, outboxMessageHandler, cancellationToken);
                 }
                 catch (RpcException e)
                 {
@@ -87,89 +86,15 @@ protected async Task ProcessMessagesAsync(CancellationToken cancellationToken =
         }
     }
 
-    async Task ProcessGroupMessagesAsync(OutboxMessage message, CancellationToken cancellationToken = default)
+    async Task ProcessGroupMessagesAsync(
+        OutboxMessage message,
+        IOutboxMessageHandler outboxMessageHandler,
+        CancellationToken cancellationToken = default
+    )
     {
-        string requestContent;
-        bool deleteMessageFromDisk = false;
-        if (message.RequestContent is null)
-        {
-            await using var requestContentStream = await _sharedFileService.OpenReadAsync(
-                $"outbox/{message.Id}",
-                cancellationToken
-            );
-            requestContent = new StreamReader(requestContentStream).ReadToEnd();
-            deleteMessageFromDisk = true;
-        }
-        else
-        {
-            requestContent = message.RequestContent;
-        }
-        switch (message.Method)
-        {
-            case OutboxMessageMethod.BuildStarted:
-                await _client.BuildStartedAsync(
-                    JsonSerializer.Deserialize<BuildStartedRequest>(requestContent),
-                    cancellationToken: cancellationToken
-                );
-                break;
-            case OutboxMessageMethod.BuildCompleted:
-                await _client.BuildCompletedAsync(
-                    JsonSerializer.Deserialize<BuildCompletedRequest>(requestContent),
-                    cancellationToken: cancellationToken
-                );
-                break;
-            case OutboxMessageMethod.BuildCanceled:
-                await _client.BuildCanceledAsync(
-                    JsonSerializer.Deserialize<BuildCanceledRequest>(requestContent),
-                    cancellationToken: cancellationToken
-                );
-                break;
-            case OutboxMessageMethod.BuildFaulted:
-                await _client.BuildFaultedAsync(
-                    JsonSerializer.Deserialize<BuildFaultedRequest>(requestContent),
-                    cancellationToken: cancellationToken
-                );
-                break;
-            case OutboxMessageMethod.BuildRestarting:
-                await _client.BuildRestartingAsync(
-                    JsonSerializer.Deserialize<BuildRestartingRequest>(requestContent),
-                    cancellationToken: cancellationToken
-                );
-                break;
-            case OutboxMessageMethod.InsertPretranslations:
-
-                {
-                    using var call = _client.InsertPretranslations(cancellationToken: cancellationToken);
-                    var requests = JsonSerializer.Deserialize<List<InsertPretranslationRequest>>(requestContent);
-                    foreach (
-                        var request in requests?.Where(r => r != null)
-                            ?? Enumerable.Empty<InsertPretranslationRequest>()
-                    )
-                    {
-                        await call.RequestStream.WriteAsync(request, cancellationToken: cancellationToken);
-                    }
-                    await call.RequestStream.CompleteAsync();
-                }
-                break;
-            case OutboxMessageMethod.IncrementTranslationEngineCorpusSize:
-                await _client.IncrementTranslationEngineCorpusSizeAsync(
-                    JsonSerializer.Deserialize<IncrementTranslationEngineCorpusSizeRequest>(requestContent),
-                    cancellationToken: cancellationToken
-                );
-                break;
-            default:
-                await _messages.DeleteAsync(message.Id);
-                _logger.LogWarning(
-                    "Unknown method: {message.Method}.  Deleting the message from the list.",
-                    message.Method.ToString()
-                );
-                break;
-        }
+        await outboxMessageHandler.SendMessageAsync(message, cancellationToken);
         await _messages.DeleteAsync(message.Id);
-        if (deleteMessageFromDisk)
-        {
-            await _sharedFileService.DeleteAsync($"outbox/{message.Id}", cancellationToken);
-        }
+        await outboxMessageHandler.CleanupMessageAsync(message, cancellationToken);
     }
 
     async Task<bool> CheckIfFinalMessageAttempt(OutboxMessage message, Exception e)