Bulk fix.

backend/src/Squidex.Domain.Apps.Entities/Contents/BulkUpdateCommandMiddleware.cs

@@ -6,6 +6,7 @@
 // ==========================================================================
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
@@ -18,6 +19,8 @@ using Squidex.Infrastructure.Translations;
 using Squidex.Shared;
 
 #pragma warning disable CA1826 // Do not use Enumerable methods on indexable collections
+#pragma warning disable SA1313 // Parameter names should begin with lower-case letter
+#pragma warning disable RECS0082 // Parameter has the same name as a member and hides it
 
 namespace Squidex.Domain.Apps.Entities.Contents
 {
@@ -26,6 +29,17 @@ namespace Squidex.Domain.Apps.Entities.Contents
         private readonly IContentQueryService contentQuery;
         private readonly IContextProvider contextProvider;
 
+        private sealed record BulkTask(
+            ICommandBus Bus,
+            Context Context,
+            string Schema,
+            BulkUpdateJob Job,
+            BulkUpdateContents Command
+        )
+        {
+            public BulkUpdateResultItem Result { get; } = new BulkUpdateResultItem();
+        }
+
         public BulkUpdateCommandMiddleware(IContentQueryService contentQuery, IContextProvider contextProvider)
         {
             Guard.NotNull(contentQuery, nameof(contentQuery));
@@ -41,50 +55,82 @@ namespace Squidex.Domain.Apps.Entities.Contents
             {
                 if (bulkUpdates.Jobs?.Length > 0)
                 {
+                    var actionBlock = new ActionBlock<BulkTask>(async task =>
+                    {
+                        try
+                        {
+                            await ExecuteTaskAsync(task);
+                        }
+                        catch (Exception ex)
+                        {
+                            task.Result.Exception = ex;
+                        }
+                    }, new ExecutionDataflowBlockOptions
+                    {
+                        MaxDegreeOfParallelism = Math.Max(1, Environment.ProcessorCount / 2)
+                    });
+
                     var requestContext = contextProvider.Context.WithoutContentEnrichment().WithUnpublished(true);
                     var requestedSchema = bulkUpdates.SchemaId.Name;
 
-                    async Task PublishAsync<TCommand>(BulkUpdateJob job, TCommand command, string permissionId) where TCommand : ContentCommand
+                    var results = new List<BulkUpdateResultItem>(bulkUpdates.Jobs.Length);
-                    {
-                        SimpleMapper.Map(bulkUpdates, command);
 
-                        if (!string.IsNullOrWhiteSpace(job.Schema))
+                    for (var i = 0; i < bulkUpdates.Jobs.Length; i++)
                     {
-                            var schema = await contentQuery.GetSchemaOrThrowAsync(requestContext, job.Schema);
+                        var task = new BulkTask(
+                            context.CommandBus,
+                            requestContext,
+                            requestedSchema,
+                            bulkUpdates.Jobs[i],
+                            bulkUpdates);
 
-                            command.SchemaId = schema.NamedId();
+                        await actionBlock.SendAsync(task);
+
+                        results.Add(task.Result);
                     }
 
-                        var permission = Permissions.ForApp(permissionId, command.AppId.Name, command.SchemaId.Name);
+                    actionBlock.Complete();
 
-                        if (!requestContext.Permissions.Allows(permission))
+                    await actionBlock.Completion;
+
+                    context.Complete(new BulkUpdateResult(results));
+                }
+                else
                 {
-                            throw new DomainForbiddenException("Forbidden");
+                    context.Complete(new BulkUpdateResult());
+                }
+            }
+            else
+            {
+                await next(context);
             }
-
-                        command.ExpectedVersion = job.ExpectedVersion;
-
-                        await context.CommandBus.PublishAsync(command);
         }
 
-                    var results = new BulkUpdateResultItem[bulkUpdates.Jobs.Length];
+        private async Task ExecuteTaskAsync(BulkTask task)
-
-                    var actionBlock = new ActionBlock<int>(async index =>
         {
-                        var job = bulkUpdates.Jobs[index];
+            var job = task.Job;
 
-                        var result = new BulkUpdateResultItem();
+            var resolvedId = await FindIdAsync(task);
 
-                        try
+            DomainId id;
-                        {
-                            var id = await FindIdAsync(requestContext, requestedSchema, job);
 
-                            if (job.Type != BulkUpdateType.Upsert && (id == null || id == DomainId.Empty))
+            if (resolvedId == null || resolvedId == DomainId.Empty)
+            {
+                if (job.Type == BulkUpdateType.Upsert)
+                {
+                    id = DomainId.NewGuid();
+                }
+                else
                 {
                     throw new DomainObjectNotFoundException("undefined");
                 }
+            }
+            else
+            {
+                id = resolvedId.Value;
+            }
 
-                            result.ContentId = id;
+            task.Result.ContentId = id;
 
             switch (job.Type)
             {
@@ -92,84 +138,70 @@ namespace Squidex.Domain.Apps.Entities.Contents
                     {
                         var command = new UpsertContent { Data = job.Data! };
 
-                                        if (id != null && id != DomainId.Empty)
+                        await PublishAsync(id, task, command, Permissions.AppContentsUpsert);
-                                        {
-                                            command.ContentId = id.Value;
-                                        }
-
-                                        result.ContentId = command.ContentId;
-
-                                        await PublishAsync(job, command, Permissions.AppContentsUpsert);
                         break;
                     }
 
                 case BulkUpdateType.Validate:
                     {
-                                        var command = new ValidateContent { ContentId = id.Value };
+                        var command = new ValidateContent();
 
-                                        await PublishAsync(job, command, Permissions.AppContentsRead);
+                        await PublishAsync(id, task, command, Permissions.AppContentsRead);
                         break;
                     }
 
                 case BulkUpdateType.ChangeStatus:
                     {
-                                        var command = new ChangeContentStatus { ContentId = id.Value, Status = job.Status };
+                        var command = new ChangeContentStatus { Status = job.Status, DueTime = job.DueTime };
 
-                                        await PublishAsync(job, command, Permissions.AppContentsUpdate);
+                        await PublishAsync(id, task, command, Permissions.AppContentsUpdate);
                         break;
                     }
 
                 case BulkUpdateType.Delete:
                     {
-                                        var command = new DeleteContent { ContentId = id.Value };
+                        var command = new DeleteContent();
 
-                                        await PublishAsync(job, command, Permissions.AppContentsDelete);
+                        await PublishAsync(id, task, command, Permissions.AppContentsDelete);
                         break;
                     }
             }
         }
-                        catch (Exception ex)
-                        {
-                            result.Exception = ex;
-                        }
 
-                        results[index] = result;
+        private async Task PublishAsync<TCommand>(DomainId id, BulkTask task, TCommand command, string permissionId) where TCommand : ContentCommand
-                    }, new ExecutionDataflowBlockOptions
         {
-                        MaxDegreeOfParallelism = Math.Max(1, Environment.ProcessorCount / 2)
+            SimpleMapper.Map(task.Command, command);
-                    });
 
-                    for (var i = 0; i < bulkUpdates.Jobs.Length; i++)
+            command.ContentId = id;
+
+            if (!string.IsNullOrWhiteSpace(task.Job.Schema))
             {
-                        await actionBlock.SendAsync(i);
+                var schema = await contentQuery.GetSchemaOrThrowAsync(task.Context, task.Schema);
-                    }
 
-                    actionBlock.Complete();
+                command.SchemaId = schema.NamedId();
+            }
 
-                    await actionBlock.Completion;
+            var permission = Permissions.ForApp(permissionId, command.AppId.Name, command.SchemaId.Name);
 
-                    context.Complete(new BulkUpdateResult(results));
+            if (!task.Context.Permissions.Allows(permission))
-                }
-                else
             {
-                    context.Complete(new BulkUpdateResult());
+                throw new DomainForbiddenException("Forbidden");
-                }
-            }
-            else
-            {
-                await next(context);
             }
+
+            command.ExpectedVersion = task.Command.ExpectedVersion;
+
+            await task.Bus.PublishAsync(command);
         }
 
-        private async Task<DomainId?> FindIdAsync(Context context, string schema, BulkUpdateJob job)
+        private async Task<DomainId?> FindIdAsync(BulkTask task)
         {
-            var id = job.Id;
+            var id = task.Job.Id;
 
-            if (id == null && job.Query != null)
+            if (id == null && task.Job.Query != null)
             {
-                job.Query.Take = 1;
+                task.Job.Query.Take = 1;
 
-                var existing = await contentQuery.QueryAsync(context, schema, Q.Empty.WithJsonQuery(job.Query));
+                var existing = await contentQuery.QueryAsync(task.Context, task.Schema, Q.Empty.WithJsonQuery(task.Job.Query));
 
                 if (existing.Total > 1)
                 {

backend/src/Squidex.Domain.Apps.Entities/Contents/BulkUpdateResultItem.cs

@@ -12,7 +12,7 @@ namespace Squidex.Domain.Apps.Entities.Contents
 {
     public sealed class BulkUpdateResultItem
     {
-        public DomainId? ContentId { get; set; }
+        public DomainId ContentId { get; set; }
 
         public Exception? Exception { get; set; }
     }

backend/src/Squidex.Domain.Apps.Entities/Contents/Commands/BulkUpdateJob.cs

@@ -5,6 +5,7 @@
 //  All rights reserved. Licensed under the MIT license.
 // ==========================================================================
 
+using NodaTime;
 using Squidex.Domain.Apps.Core.Contents;
 using Squidex.Infrastructure;
 using Squidex.Infrastructure.Json.Objects;
@@ -20,6 +21,8 @@ namespace Squidex.Domain.Apps.Entities.Contents.Commands
 
         public Status Status { get; set; }
 
+        public Instant? DueTime { get; set; }
+
         public BulkUpdateType Type { get; set; }
 
         public NamedContentData? Data { get; set; }

backend/src/Squidex/Areas/Api/Controllers/Contents/Models/BulkResultDto.cs

@@ -20,9 +20,9 @@ namespace Squidex.Areas.Api.Controllers.Contents.Models
         public ErrorDto? Error { get; set; }
 
         /// <summary>
-        /// The id of the content when the import succeeds.
+        /// The id of the content that has been handled successfully or not.
         /// </summary>
-        public DomainId? ContentId { get; set; }
+        public DomainId ContentId { get; set; }
 
         public static BulkResultDto FromImportResult(BulkUpdateResultItem result, HttpContext httpContext)
         {

backend/src/Squidex/Areas/Api/Controllers/Contents/Models/BulkUpdateJobDto.cs

@@ -5,6 +5,7 @@
 //  All rights reserved. Licensed under the MIT license.
 // ==========================================================================
 
+using NodaTime;
 using Squidex.Domain.Apps.Core.Contents;
 using Squidex.Domain.Apps.Entities.Contents.Commands;
 using Squidex.Infrastructure;
@@ -36,6 +37,11 @@ namespace Squidex.Areas.Api.Controllers.Contents.Models
         /// </summary>
         public Status Status { get; set; }
 
+        /// <summary>
+        /// The due time.
+        /// </summary>
+        public Instant? DueTime { get; set; }
+
         /// <summary>
         /// The update type.
         /// </summary>

backend/tests/Squidex.Domain.Apps.Entities.Tests/Contents/BulkUpdateCommandMiddlewareTests.cs

@@ -9,6 +9,7 @@ using System;
 using System.Security.Claims;
 using System.Threading.Tasks;
 using FakeItEasy;
+using NodaTime;
 using Squidex.Domain.Apps.Core.Contents;
 using Squidex.Domain.Apps.Entities.Contents.Commands;
 using Squidex.Infrastructure;
@@ -198,7 +199,26 @@ namespace Squidex.Domain.Apps.Entities.Contents
 
             Assert.Single(result, x => x.ContentId == id && x.Exception == null);
 
-            A.CallTo(() => commandBus.PublishAsync(A<ChangeContentStatus>.That.Matches(x => x.ContentId == id)))
+            A.CallTo(() => commandBus.PublishAsync(A<ChangeContentStatus>.That.Matches(x => x.ContentId == id && x.DueTime == null)))
+                .MustHaveHappened();
+        }
+
+        [Fact]
+        public async Task Should_change_content_status_with_due_time()
+        {
+            SetupContext(Permissions.AppContentsUpdate);
+
+            var time = Instant.FromDateTimeUtc(DateTime.UtcNow);
+
+            var (id, _, _) = CreateTestData(false);
+
+            var command = BulkCommand(BulkUpdateType.ChangeStatus, id: id, dueTime: time);
+
+            var result = await PublishAsync(command);
+
+            Assert.Single(result, x => x.ContentId == id && x.Exception == null);
+
+            A.CallTo(() => commandBus.PublishAsync(A<ChangeContentStatus>.That.Matches(x => x.ContentId == id && x.DueTime == time)))
                 .MustHaveHappened();
         }
 
@@ -298,14 +318,22 @@ namespace Squidex.Domain.Apps.Entities.Contents
             return (context.PlainResult as BulkUpdateResult)!;
         }
 
-        private BulkUpdateContents BulkCommand(BulkUpdateType type, Query<IJsonValue>? query = null, DomainId? id = null, NamedContentData? data = null)
+        private BulkUpdateContents BulkCommand(BulkUpdateType type, Query<IJsonValue>? query = null,
+            DomainId? id = null, NamedContentData? data = null, Instant? dueTime = null)
         {
             return new BulkUpdateContents
             {
                 AppId = appId,
                 Jobs = new[]
                 {
-                    new BulkUpdateJob { Type = type, Query = query, Id = id, Data = data! }
+                    new BulkUpdateJob
+                    {
+                        Type = type,
+                        Id = id,
+                        Data = data!,
+                        DueTime = dueTime,
+                        Query = query,
+                    }
                 },
                 SchemaId = schemaId
             };