Ensure json is streamed from the reqest into the db (#1608)

* import commits via a temp table and then merge data into the commits table

* define and use a streaming json IAsyncEnumerable model

Author: hahn-kev

backend/FwLite/FwLiteWeb/Routes/ProjectRoutes.cs

@@ -42,6 +42,7 @@ public static IEndpointConventionBuilder MapProjectRoutes(this WebApplication ap
             async (SyncService syncService,
                 IOptions<AuthConfig> options,
                 string serverAuthority,
+                [FromRoute(Name = CrdtMiniLcmApiHub.ProjectRouteKey)]string project,
                 [FromQuery] Guid lexboxProjectId) =>
             {
                 var server = options.Value.GetServerByAuthority(serverAuthority);

backend/LexBoxApi/Controllers/CrdtController.cs

@@ -4,6 +4,7 @@
 using LexBoxApi.Auth.Attributes;
 using LexBoxApi.GraphQL;
 using LexBoxApi.Hub;
+using LexBoxApi.Models;
 using LexBoxApi.Services;
 using LexCore.Auth;
 using LexCore.Entities;
@@ -12,6 +13,7 @@
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Options;
 using MiniLcm.Push;
 
 namespace LexBoxApi.Controllers;
@@ -27,7 +29,8 @@ public class CrdtController(
     LoggedInContext loggedInContext,
     ProjectService projectService,
     CrdtCommitService crdtCommitService,
-    LexAuthService lexAuthService) : ControllerBase
+    LexAuthService lexAuthService
+    ) : ControllerBase
 {
     [HttpGet("{projectId}/get")]
     public async Task<ActionResult<SyncState>> GetSyncState(Guid projectId)
@@ -37,10 +40,11 @@ public async Task<ActionResult<SyncState>> GetSyncState(Guid projectId)
     }
 
     [HttpPost("{projectId}/add")]
-    public async Task<ActionResult> Add(Guid projectId, [FromBody] IAsyncEnumerable<ServerCommit> commits, Guid? clientId)
+
+    public async Task<ActionResult> Add(Guid projectId, StreamJsonAsyncEnumerable<ServerCommit> commits, Guid? clientId, CancellationToken token)
     {
         await permissionService.AssertCanSyncProject(projectId);
-        await crdtCommitService.AddCommits(projectId, commits);
+        await crdtCommitService.AddCommits(projectId, commits, token);
 
         await hubContext.Clients.Group(CrdtProjectChangeHub.ProjectGroup(projectId)).OnProjectUpdated(projectId, clientId);
         return Ok();

backend/LexBoxApi/Models/StreamJsonAsyncEnumerable.cs

@@ -0,0 +1,57 @@
+using System.Reflection;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.ModelBinding;
+using Microsoft.Extensions.Options;
+
+namespace LexBoxApi.Models;
+
+/// <summary>
+/// by default if you set IAsyncEnumerable<T> as a parameter type, it will be created using JsonSerializer.DeserializeAsync
+/// this will buffer the data into a list, rather than streaming it. This class exists to allow streaming
+/// </summary>
+[UseStreamingBinder]
+public class StreamJsonAsyncEnumerable<T>(IAsyncEnumerable<T> stream): IAsyncEnumerable<T>
+{
+    public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = new CancellationToken())
+    {
+        return stream.GetAsyncEnumerator(cancellationToken);
+    }
+}
+
+internal class UseStreamingBinderAttribute() : ModelBinderAttribute(typeof(StreamJsonAsyncEnumerableConverter))
+{
+    public override BindingSource? BindingSource => BindingSource.Body;
+}
+public class StreamJsonAsyncEnumerableConverter(ILogger<StreamJsonAsyncEnumerableConverter> logger): IModelBinder
+{
+    private static readonly MethodInfo BindMethod = new Action<ModelBindingContext>(Bind<object>).Method.GetGenericMethodDefinition();
+    public Task BindModelAsync(ModelBindingContext bindingContext)
+    {
+        if (!bindingContext.ModelType.IsGenericType)
+        {
+            throw new InvalidOperationException("Model type must be generic");
+        }
+        try
+        {
+            var genericType = bindingContext.ModelType.GetGenericArguments()[0];
+            var bindMethodGeneric = BindMethod.MakeGenericMethod(genericType);
+            bindMethodGeneric.Invoke(null, [bindingContext]);
+        }
+        catch (Exception e)
+        {
+            logger.LogError(e, "problem deserializing json");
+            bindingContext.ModelState.TryAddModelException("this", e);
+        }
+        return Task.CompletedTask;
+    }
+
+    private static void Bind<T>(ModelBindingContext bindingContext) where T : class
+    {
+        var options = bindingContext.HttpContext.RequestServices.GetRequiredService<IOptions<JsonOptions>>().Value
+            .JsonSerializerOptions;
+        var enumerable = JsonSerializer.DeserializeAsyncEnumerable<T>(bindingContext.HttpContext.Request.BodyReader.AsStream(), options, bindingContext.HttpContext.RequestAborted);
+        bindingContext.Result = ModelBindingResult.Success(new StreamJsonAsyncEnumerable<T>(enumerable.OfType<T>()));
+    }
+}

backend/LexBoxApi/Services/CrdtCommitService.cs

@@ -1,6 +1,7 @@
 using LexCore.Utils;
 using LexData;
 using LinqToDB;
+using LinqToDB.Data;
 using LinqToDB.EntityFrameworkCore;
 using LinqToDB.EntityFrameworkCore.Internal;
 using SIL.Harmony.Core;
@@ -9,53 +10,34 @@ namespace LexBoxApi.Services;
 
 public class CrdtCommitService(LexBoxDbContext dbContext)
 {
-    public async Task AddCommits(Guid projectId, IAsyncEnumerable<ServerCommit> commits)
+    public async Task AddCommits(Guid projectId, IAsyncEnumerable<ServerCommit> commits, CancellationToken token = default)
     {
-        await using var transaction = await dbContext.Database.BeginTransactionAsync();
-        var commitsTable = dbContext.CreateLinqToDBContext().GetTable<ServerCommit>();
-        const int commitsThreshold = 100;
-        const int changeThreshold = 150;
-        var currentChangeCount = 0;
-        var commitBucket = new List<ServerCommit>(100);
-        await foreach (var serverCommit in commits)
-        {
-            commitBucket.Add(serverCommit);
-            currentChangeCount += serverCommit.ChangeEntities.Count;
-            if (currentChangeCount < changeThreshold && commitBucket.Count < commitsThreshold)
+        await using var transaction = await dbContext.Database.BeginTransactionAsync(token);
+        var linqToDbContext = dbContext.CreateLinqToDBContext();
+        await using var tmpTable = await linqToDbContext.CreateTempTableAsync<ServerCommit>($"tmp_crdt_commit_import_{projectId}__{Guid.NewGuid()}", cancellationToken: token);
+        await tmpTable.BulkCopyAsync(new BulkCopyOptions{BulkCopyType = BulkCopyType.ProviderSpecific, MaxBatchSize = 10}, commits, token);
+
+        var commitsTable = linqToDbContext.GetTable<ServerCommit>();
+        await commitsTable
+            .Merge()
+            .Using(tmpTable)
+            .OnTargetKey()
+            .InsertWhenNotMatched(commit => new ServerCommit(commit.Id)
             {
-                continue;
-            }
-
-            await FlushCommits();
-        }
-
-        if (commitBucket.Count > 0) await FlushCommits();
-
-        await transaction.CommitAsync();
-
-        async Task FlushCommits()
-        {
-            await commitsTable
-                .Merge()
-                .Using(commitBucket)
-                .OnTargetKey()
-                .InsertWhenNotMatched(commit => new ServerCommit(commit.Id)
+                Id = commit.Id,
+                ClientId = commit.ClientId,
+                HybridDateTime = new HybridDateTime(commit.HybridDateTime.DateTime, commit.HybridDateTime.Counter)
                 {
-                    Id = commit.Id,
-                    ClientId = commit.ClientId,
-                    HybridDateTime = new HybridDateTime(commit.HybridDateTime.DateTime, commit.HybridDateTime.Counter)
-                    {
-                        DateTime = commit.HybridDateTime.DateTime, Counter = commit.HybridDateTime.Counter
-                    },
-                    ProjectId = projectId,
-                    Metadata = commit.Metadata,
-                    //without this sql cast the value will be treated as text and fail to insert into the jsonb column
-                    ChangeEntities = Sql.Expr<List<ChangeEntity<ServerJsonChange>>>($"{commit.ChangeEntities}::jsonb")
-                })
-                .MergeAsync();
-            commitBucket.Clear();
-            currentChangeCount = 0;
-        }
+                    DateTime = commit.HybridDateTime.DateTime, Counter = commit.HybridDateTime.Counter
+                },
+                ProjectId = projectId,
+                Metadata = commit.Metadata,
+                //without this sql cast the value will be treated as text and fail to insert into the jsonb column
+                ChangeEntities = Sql.Expr<List<ChangeEntity<ServerJsonChange>>>($"{commit.ChangeEntities}::jsonb")
+            })
+            .MergeAsync(token);
+
+        await transaction.CommitAsync(token);
     }
 
     public IAsyncEnumerable<ServerCommit> GetMissingCommits(Guid projectId, SyncState localState, SyncState remoteState)