diff --git a/src/protagonist/API.Tests/Integration/CustomerAdjunctQueueTests.cs b/src/protagonist/API.Tests/Integration/CustomerAdjunctQueueTests.cs index a153461fc..acb253546 100644 --- a/src/protagonist/API.Tests/Integration/CustomerAdjunctQueueTests.cs +++ b/src/protagonist/API.Tests/Integration/CustomerAdjunctQueueTests.cs @@ -11,6 +11,7 @@ using DLCS.AWS.SNS.Messaging; using DLCS.Model.Assets; using DLCS.Model.Messaging; +using DLCS.Model.Processing; using DLCS.Repository; using DLCS.Web.Response; using FakeItEasy; @@ -23,6 +24,7 @@ using Test.Helpers.Integration; using Test.Helpers.Integration.Infrastructure; using AdjunctBatch = DLCS.HydraModel.AdjunctBatch; +using CustomerAdjunctQueue = DLCS.HydraModel.CustomerAdjunctQueue; namespace API.Tests.Integration; @@ -420,6 +422,230 @@ public async Task PostAdjunctBatch_Returns201_AdjunctHasBatchIdSet() adjunct.Batch.Should().Be(batchId, "adjunct should reference the created batch"); } + [Fact] + public async Task GetAdjunctQueue_Returns404_WhenQueueNotFound() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.NotFound); + } + + [Fact] + public async Task GetAdjunctQueue_Returns200_WithSizeFromQueueRow_WhenNoBatches() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + await dbContext.Queues.AddAsync(new Queue { Customer = assetId.Customer, Name = QueueNames.Adjunct, Size = 7 }); + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var queue = await response.ReadAsHydraResponseAsync(); + queue.Size.Should().Be(7); + queue.BatchesWaiting.Should().Be(0); + queue.AdjunctsWaiting.Should().Be(0); + } + + [Fact] + public async Task GetAdjunctQueue_Returns200_WithBatchesWaitingAndAdjunctsWaiting_ExcludingInProgressAndFinishedBatches() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + await dbContext.Queues.AddAsync(new Queue { Customer = assetId.Customer, Name = QueueNames.Adjunct, Size = 10 }); + // finished - excluded + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, count: 5, completed: 5, + finished: DateTime.UtcNow); + // unfinished, nothing started yet - counts as waiting + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, count: 5, completed: 0); + // unfinished but already partially processed - excluded, still being worked on + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, count: 8, completed: 6); + // unfinished but has an error recorded - also excluded, platform has started working on it + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, count: 3, errors: 1); + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var queue = await response.ReadAsHydraResponseAsync(); + queue.Size.Should().Be(10); + queue.BatchesWaiting.Should().Be(1, "only batch 2 has not started processing"); + queue.AdjunctsWaiting.Should().Be(5, "only batch 2's 5 adjuncts have not started processing"); + } + + [Fact] + public async Task GetAdjunctQueue_Links_ResolveToWorkingCollectionEndpoints() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + await dbContext.Queues.AddAsync(new Queue { Customer = assetId.Customer, Name = QueueNames.Adjunct, Size = 0 }); + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, count: 1, completed: 0); + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, count: 1, completed: 1, + finished: DateTime.UtcNow); + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue"); + response.StatusCode.Should().Be(HttpStatusCode.OK); + var queue = await response.ReadAsHydraResponseAsync(); + + // Assert + queue.Batches.Should().EndWith("/adjunctQueue/batches"); + queue.Active.Should().EndWith("/adjunctQueue/active"); + queue.Recent.Should().EndWith("/adjunctQueue/recent"); + + var batches = await (await httpClient.AsCustomer(assetId.Customer).GetAsync(queue.Batches)) + .ReadAsHydraResponseAsync>(); + batches.TotalItems.Should().Be(2); + + var active = await (await httpClient.AsCustomer(assetId.Customer).GetAsync(queue.Active)) + .ReadAsHydraResponseAsync>(); + active.TotalItems.Should().Be(1); + + var recent = await (await httpClient.AsCustomer(assetId.Customer).GetAsync(queue.Recent)) + .ReadAsHydraResponseAsync>(); + recent.TotalItems.Should().Be(1); + } + + [Fact] + public async Task GetAdjunctBatches_Returns200_Empty_WhenNoBatches() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue/batches"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var batches = await response.ReadAsHydraResponseAsync>(); + batches.Members.Should().BeEmpty(); + } + + [Fact] + public async Task GetAdjunctBatches_Returns200_MostRecentlySubmittedFirst_ByDefault() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + var earlier = DateTime.UtcNow.AddMinutes(-10); + var later = DateTime.UtcNow; + var earlierBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + submitted: earlier, finished: DateTime.UtcNow)).Entity; + var laterBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + submitted: later)).Entity; + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue/batches"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var batches = await response.ReadAsHydraResponseAsync>(); + batches.Members.Select(b => b.GetLastPathElementAsInt()).Should().ContainInOrder(laterBatch.Id, earlierBatch.Id); + } + + [Fact] + public async Task GetAdjunctBatches_Returns200_OrdersAscending_WhenOrderByRequested() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + var earlier = DateTime.UtcNow.AddMinutes(-10); + var later = DateTime.UtcNow; + var earlierBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + submitted: earlier)).Entity; + var laterBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + submitted: later)).Entity; + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue/batches?orderBy=submitted"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var batches = await response.ReadAsHydraResponseAsync>(); + batches.Members.Select(b => b.GetLastPathElementAsInt()).Should().ContainInOrder(earlierBatch.Id, laterBatch.Id); + } + + [Fact] + public async Task GetActiveAdjunctBatches_Returns200_OnlyUnfinishedBatches() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + var activeBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer)).Entity; + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, finished: DateTime.UtcNow); + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue/active"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var batches = await response.ReadAsHydraResponseAsync>(); + batches.Members.Should().ContainSingle(b => b.GetLastPathElementAsInt() == activeBatch.Id); + } + + [Fact] + public async Task GetActiveAdjunctBatches_Returns200_OrdersDescending_WhenOrderByDescendingRequested() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + var earlier = DateTime.UtcNow.AddMinutes(-10); + var later = DateTime.UtcNow; + var earlierBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + submitted: earlier)).Entity; + var laterBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + submitted: later)).Entity; + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue/active?orderByDescending=submitted"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var batches = await response.ReadAsHydraResponseAsync>(); + batches.Members.Select(b => b.GetLastPathElementAsInt()).Should().ContainInOrder(laterBatch.Id, earlierBatch.Id); + } + + [Fact] + public async Task GetRecentAdjunctBatches_Returns200_OnlyFinishedBatches_OrderedByFinishedDescending() + { + // Arrange + var assetId = AssetIdGenerator.GetAssetId(); + await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer); + var olderFinishedBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + finished: DateTime.UtcNow.AddDays(-7))).Entity; + var newerFinishedBatch = (await dbContext.AdjunctBatches.AddTestAdjunctBatch(customer: assetId.Customer, + finished: DateTime.UtcNow)).Entity; + await dbContext.SaveChangesAsync(); + + // Act + var response = await httpClient.AsCustomer(assetId.Customer) + .GetAsync($"/customers/{assetId.Customer}/adjunctQueue/recent"); + + // Assert + response.StatusCode.Should().Be(HttpStatusCode.OK); + var batches = await response.ReadAsHydraResponseAsync>(); + batches.Members.Select(b => b.GetLastPathElementAsInt()).Should().ContainInOrder(newerFinishedBatch.Id, olderFinishedBatch.Id); + } + [Fact] public async Task GetAdjunctBatch_Returns404_WhenBatchNotFound() { diff --git a/src/protagonist/API/Features/AdjunctQueues/AdjunctQueryX.cs b/src/protagonist/API/Features/AdjunctQueues/AdjunctQueryX.cs index 9badcefe4..0032a98d9 100644 --- a/src/protagonist/API/Features/AdjunctQueues/AdjunctQueryX.cs +++ b/src/protagonist/API/Features/AdjunctQueues/AdjunctQueryX.cs @@ -16,4 +16,13 @@ public static IQueryable AsOrderedAdjunctQuery(this IQueryable => orderable.Descending ? query.OrderByDescending(a => a.Created) : query.OrderBy(a => a.Created); + + /// + /// Order adjunct batches by Submitted date - currently the only supported orderBy field for adjunct batches. + /// Any other/unrecognised orderBy value is silently ignored and Submitted is used. + /// + public static IQueryable AsOrderedAdjunctBatchQuery(this IQueryable query, IOrderableRequest orderable) + => orderable.Descending + ? query.OrderByDescending(b => b.Submitted) + : query.OrderBy(b => b.Submitted); } diff --git a/src/protagonist/API/Features/AdjunctQueues/Converters/AdjunctQueueConverter.cs b/src/protagonist/API/Features/AdjunctQueues/Converters/AdjunctQueueConverter.cs new file mode 100644 index 000000000..553c06b68 --- /dev/null +++ b/src/protagonist/API/Features/AdjunctQueues/Converters/AdjunctQueueConverter.cs @@ -0,0 +1,23 @@ +using HydraAdjunctQueue = DLCS.HydraModel.CustomerAdjunctQueue; +using EntityAdjunctQueue = DLCS.Model.Processing.AdjunctQueue; + +namespace API.Features.AdjunctQueues.Converters; + +/// +/// Conversion between API and EF forms of AdjunctQueue resource +/// +public static class AdjunctQueueConverter +{ + /// + /// Convert AdjunctQueue entity to API resource + /// + public static HydraAdjunctQueue ToHydra(this EntityAdjunctQueue adjunctQueue, string baseUrl) + { + var hydra = new HydraAdjunctQueue(baseUrl, adjunctQueue.Customer); + hydra.Size = adjunctQueue.Size; + hydra.BatchesWaiting = adjunctQueue.BatchesWaiting; + hydra.AdjunctsWaiting = adjunctQueue.AdjunctsWaiting; + + return hydra; + } +} diff --git a/src/protagonist/API/Features/AdjunctQueues/CustomerAdjunctQueueController.cs b/src/protagonist/API/Features/AdjunctQueues/CustomerAdjunctQueueController.cs index a4c9c4ce9..e1170a2e9 100644 --- a/src/protagonist/API/Features/AdjunctQueues/CustomerAdjunctQueueController.cs +++ b/src/protagonist/API/Features/AdjunctQueues/CustomerAdjunctQueueController.cs @@ -24,6 +24,86 @@ public class CustomerAdjunctQueueController( IMediator mediator) : HydraController(options.Value, mediator) { + /// + /// Get details of default customer adjunct queue + /// + /// Id of customer to get adjunct queue details for + /// Current cancellation token + /// Hydra JSON-LD CustomerAdjunctQueue object + [HttpGet] + [ProducesResponseType(StatusCodes.Status200OK, Type = typeof(CustomerAdjunctQueue))] + [ProducesResponseType(StatusCodes.Status404NotFound)] + public async Task GetCustomerAdjunctQueue([FromRoute] int customerId, + CancellationToken cancellationToken) + { + return await HandleFetch( + new GetCustomerAdjunctQueue(customerId), + queue => queue.ToHydra(GetUrlRoots().BaseUrl), + errorTitle: "Get Customer Adjunct Queue failed", + cancellationToken: cancellationToken); + } + + /// + /// Get details of all customer adjunct batches. + /// + /// Supports ?page= and ?pageSize= query parameters for paging + /// + /// Id of customer + /// Current cancellation token + /// Hydra JSON-LD collection of AdjunctBatch objects + [HttpGet] + [Route("batches")] + [ProducesResponseType(StatusCodes.Status200OK, Type = typeof(HydraCollection))] + public async Task GetAdjunctBatches([FromRoute] int customerId, CancellationToken cancellationToken) + { + return await HandlePagedFetch( + new GetAdjunctBatches(customerId), + batch => batch.ToHydra(GetUrlRoots().BaseUrl), + errorTitle: "Get adjunct batches failed", + cancellationToken: cancellationToken); + } + + /// + /// Get details of customer active adjunct batches. An "active" batch is one that is incomplete. + /// + /// Supports ?page= and ?pageSize= query parameters for paging + /// + /// Id of customer + /// Current cancellation token + /// Hydra JSON-LD collection of AdjunctBatch objects + [HttpGet] + [Route("active")] + [ProducesResponseType(StatusCodes.Status200OK, Type = typeof(HydraCollection))] + public async Task GetActiveAdjunctBatches([FromRoute] int customerId, CancellationToken cancellationToken) + { + return await HandlePagedFetch( + new GetActiveAdjunctBatches(customerId), + batch => batch.ToHydra(GetUrlRoots().BaseUrl), + errorTitle: "Get active adjunct batches failed", + cancellationToken: cancellationToken); + } + + /// + /// Get details of customer recent adjunct batches. These are all batches that are finished, ordered by latest + /// finished. + /// + /// Supports ?page= and ?pageSize= query parameters for paging + /// + /// Id of customer + /// Current cancellation token + /// Hydra JSON-LD collection of AdjunctBatch objects + [HttpGet] + [Route("recent")] + [ProducesResponseType(StatusCodes.Status200OK, Type = typeof(HydraCollection))] + public async Task GetRecentAdjunctBatches([FromRoute] int customerId, CancellationToken cancellationToken) + { + return await HandlePagedFetch( + new GetRecentAdjunctBatches(customerId), + batch => batch.ToHydra(GetUrlRoots().BaseUrl), + errorTitle: "Get recent adjunct batches failed", + cancellationToken: cancellationToken); + } + /// /// Get details of specified adjunct batch. /// diff --git a/src/protagonist/API/Features/AdjunctQueues/Requests/GetActiveAdjunctBatches.cs b/src/protagonist/API/Features/AdjunctQueues/Requests/GetActiveAdjunctBatches.cs new file mode 100644 index 000000000..596d7fd79 --- /dev/null +++ b/src/protagonist/API/Features/AdjunctQueues/Requests/GetActiveAdjunctBatches.cs @@ -0,0 +1,36 @@ +using API.Infrastructure.Page; +using API.Infrastructure.Requests; +using DLCS.Model.Assets; +using DLCS.Repository; +using MediatR; +using Microsoft.EntityFrameworkCore; + +namespace API.Features.AdjunctQueues.Requests; + +/// +/// Get a paged list of all Active (incomplete) adjunct batches for customer +/// +public class GetActiveAdjunctBatches(int customerId) : IRequest>>, IPagedRequest, + IOrderableRequest +{ + public int CustomerId { get; } = customerId; + public int Page { get; set; } + public int PageSize { get; set; } + public string? Field { get; set; } + public bool Descending { get; set; } +} + +public class GetActiveAdjunctBatchesHandler(DlcsContext dlcsContext) + : IRequestHandler>> +{ + public async Task>> Handle(GetActiveAdjunctBatches request, + CancellationToken cancellationToken) + { + var result = await dlcsContext.AdjunctBatches.AsNoTracking().CreatePagedResult(request, + q => q.Where(b => b.Customer == request.CustomerId && b.Finished == null), + batches => batches.AsOrderedAdjunctBatchQuery(request), + cancellationToken: cancellationToken); + + return FetchEntityResult>.Success(result); + } +} diff --git a/src/protagonist/API/Features/AdjunctQueues/Requests/GetAdjunctBatches.cs b/src/protagonist/API/Features/AdjunctQueues/Requests/GetAdjunctBatches.cs new file mode 100644 index 000000000..cfe18001b --- /dev/null +++ b/src/protagonist/API/Features/AdjunctQueues/Requests/GetAdjunctBatches.cs @@ -0,0 +1,36 @@ +using API.Infrastructure.Page; +using API.Infrastructure.Requests; +using DLCS.Model.Assets; +using DLCS.Repository; +using MediatR; +using Microsoft.EntityFrameworkCore; + +namespace API.Features.AdjunctQueues.Requests; + +/// +/// Get a paged list of all adjunct batches for customer, most recently submitted first by default +/// +public class GetAdjunctBatches(int customerId) : IRequest>>, IPagedRequest, + IOrderableRequest +{ + public int CustomerId { get; } = customerId; + public int Page { get; set; } + public int PageSize { get; set; } + public string? Field { get; set; } + public bool Descending { get; set; } = true; +} + +public class GetAdjunctBatchesHandler(DlcsContext dlcsContext) + : IRequestHandler>> +{ + public async Task>> Handle(GetAdjunctBatches request, + CancellationToken cancellationToken) + { + var result = await dlcsContext.AdjunctBatches.AsNoTracking().CreatePagedResult(request, + q => q.Where(b => b.Customer == request.CustomerId), + batches => batches.AsOrderedAdjunctBatchQuery(request), + cancellationToken: cancellationToken); + + return FetchEntityResult>.Success(result); + } +} diff --git a/src/protagonist/API/Features/AdjunctQueues/Requests/GetCustomerAdjunctQueue.cs b/src/protagonist/API/Features/AdjunctQueues/Requests/GetCustomerAdjunctQueue.cs new file mode 100644 index 000000000..06b357968 --- /dev/null +++ b/src/protagonist/API/Features/AdjunctQueues/Requests/GetCustomerAdjunctQueue.cs @@ -0,0 +1,27 @@ +using API.Infrastructure.Requests; +using DLCS.Model.Processing; +using MediatR; + +namespace API.Features.AdjunctQueues.Requests; + +/// +/// Get details of customer adjunct queue +/// +public class GetCustomerAdjunctQueue(int customerId) : IRequest> +{ + public int CustomerId { get; } = customerId; +} + +public class GetCustomerAdjunctQueueHandler(ICustomerQueueRepository customerQueueRepository) + : IRequestHandler> +{ + public async Task> Handle(GetCustomerAdjunctQueue request, + CancellationToken cancellationToken) + { + var adjunctQueue = await customerQueueRepository.GetAdjunctQueue(request.CustomerId, cancellationToken); + + return adjunctQueue == null + ? FetchEntityResult.NotFound() + : FetchEntityResult.Success(adjunctQueue); + } +} diff --git a/src/protagonist/API/Features/AdjunctQueues/Requests/GetRecentAdjunctBatches.cs b/src/protagonist/API/Features/AdjunctQueues/Requests/GetRecentAdjunctBatches.cs new file mode 100644 index 000000000..b71c9bd3e --- /dev/null +++ b/src/protagonist/API/Features/AdjunctQueues/Requests/GetRecentAdjunctBatches.cs @@ -0,0 +1,34 @@ +using API.Infrastructure.Page; +using API.Infrastructure.Requests; +using DLCS.Model.Assets; +using DLCS.Repository; +using MediatR; +using Microsoft.EntityFrameworkCore; + +namespace API.Features.AdjunctQueues.Requests; + +/// +/// Get a paged list of all Recent (finished, ordered by finished DESC) adjunct batches for customer +/// +public class GetRecentAdjunctBatches(int customerId) : IRequest>>, IPagedRequest +{ + public int CustomerId { get; } = customerId; + public int Page { get; set; } + public int PageSize { get; set; } +} + +public class GetRecentAdjunctBatchesHandler(DlcsContext dlcsContext) + : IRequestHandler>> +{ + public async Task>> Handle(GetRecentAdjunctBatches request, + CancellationToken cancellationToken) + { + var result = await dlcsContext.AdjunctBatches.AsNoTracking().CreatePagedResult( + request, + q => q.Where(b => b.Customer == request.CustomerId && b.Finished != null), + batches => batches.OrderByDescending(b => b.Finished), + cancellationToken: cancellationToken); + + return FetchEntityResult>.Success(result); + } +} diff --git a/src/protagonist/DLCS.HydraModel/CustomerAdjunctQueue.cs b/src/protagonist/DLCS.HydraModel/CustomerAdjunctQueue.cs new file mode 100644 index 000000000..88ca52703 --- /dev/null +++ b/src/protagonist/DLCS.HydraModel/CustomerAdjunctQueue.cs @@ -0,0 +1,103 @@ +using Hydra; +using Hydra.Model; +using Newtonsoft.Json; + +namespace DLCS.HydraModel; + +[HydraClass(typeof(CustomerAdjunctQueueClass), + Description = "The customer adjunct queue resource represents an overview of outstanding adjuncts and adjunct batches.", + UriTemplate = "/customers/{0}/adjunctQueue")] +public class CustomerAdjunctQueue : DlcsResource +{ + public CustomerAdjunctQueue() + { + } + + public CustomerAdjunctQueue(string baseUrl, int customerId) + { + CustomerId = customerId; + Init(baseUrl, true, CustomerId); + } + + [JsonIgnore] + public int CustomerId { get; set; } + + [RdfProperty(Description = "Number of total adjuncts in your queue, across all batches", + Range = Names.XmlSchema.NonNegativeInteger, ReadOnly = true, WriteOnly = false)] + [JsonProperty(Order = 11, PropertyName = "size")] + public int Size { get; set; } + + [RdfProperty(Description = "Number of total adjunct batches that have not been started in your queue", + Range = Names.XmlSchema.NonNegativeInteger, ReadOnly = true, WriteOnly = false)] + [JsonProperty(Order = 12, PropertyName = "batchesWaiting")] + public long BatchesWaiting { get; set; } + + [RdfProperty(Description = "Number of adjuncts waiting to be processed. These are adjuncts that have been " + + "submitted but the platform has not yet started working on them; excludes " + + "adjuncts currently being processed", + Range = Names.XmlSchema.NonNegativeInteger, ReadOnly = true, WriteOnly = false)] + [JsonProperty(Order = 13, PropertyName = "adjunctsWaiting")] + public long AdjunctsWaiting { get; set; } + + // Hydra Link properties + [HydraLink(Description = "All adjunct batches for customer", Range = "vocab:AdjunctBatch", ReadOnly = true, WriteOnly = false)] + [JsonProperty(Order = 20, PropertyName = "batches")] + public string? Batches { get; set; } + + [HydraLink(Description = "All active adjunct batches for customer", Range = "vocab:AdjunctQueue", ReadOnly = true, WriteOnly = false)] + [JsonProperty(Order = 21, PropertyName = "active")] + public string? Active { get; set; } + + [HydraLink(Description = "All recent adjunct batches for customer", Range = "vocab:AdjunctQueue", ReadOnly = true, WriteOnly = false)] + [JsonProperty(Order = 22, PropertyName = "recent")] + public string? Recent { get; set; } +} + +public class CustomerAdjunctQueueClass : Class +{ + const string operationId = "_:customer_adjunct_queue_"; + + public CustomerAdjunctQueueClass() + { + BootstrapViaReflection(typeof(CustomerAdjunctQueue)); + } + + public override void DefineOperations() + { + SupportedOperations = CommonOperations.GetStandardResourceOperations( + operationId, "CustomerAdjunctQueue", Id, "GET", "POST"); + + GetHydraLinkProperty("batches").SupportedOperations = new[] + { + new Operation + { + Id = "_:customer_adjunct_queue_batch_collection_retrieve", + Method = "GET", + Label = "Retrieves all adjunct batches for customer", + Returns = Names.Hydra.Collection + } + }; + + GetHydraLinkProperty("active").SupportedOperations = new[] + { + new Operation + { + Id = "_:customer_adjunct_queue_active_collection_retrieve", + Method = "GET", + Label = "Retrieves the customer's currently running adjunct batches.", + Returns = Names.Hydra.Collection + } + }; + + GetHydraLinkProperty("recent").SupportedOperations = new[] + { + new Operation + { + Id = "_:customer_adjunct_queue_recent_collection_retrieve", + Method = "GET", + Label = "Retrieves the recent (finished) adjunct batches for customer.", + Returns = Names.Hydra.Collection + } + }; + } +} diff --git a/src/protagonist/DLCS.Model/Processing/AdjunctQueue.cs b/src/protagonist/DLCS.Model/Processing/AdjunctQueue.cs new file mode 100644 index 000000000..ef1b34b79 --- /dev/null +++ b/src/protagonist/DLCS.Model/Processing/AdjunctQueue.cs @@ -0,0 +1,9 @@ +namespace DLCS.Model.Processing; + +public class AdjunctQueue +{ + public int Customer { get; set; } + public int Size { get; set; } + public long BatchesWaiting { get; set; } + public long AdjunctsWaiting { get; set; } +} diff --git a/src/protagonist/DLCS.Model/Processing/ICustomerQueueRepository.cs b/src/protagonist/DLCS.Model/Processing/ICustomerQueueRepository.cs index 818b68e09..f91e6c0d9 100644 --- a/src/protagonist/DLCS.Model/Processing/ICustomerQueueRepository.cs +++ b/src/protagonist/DLCS.Model/Processing/ICustomerQueueRepository.cs @@ -11,6 +11,14 @@ public interface ICustomerQueueRepository /// Task Get(int customer, string name, CancellationToken cancellationToken); + /// + /// Get object for specified customer. + /// This consists of values from both the Queue and AdjunctBatch tables. BatchesWaiting/AdjunctsWaiting only + /// count batches that have not yet started processing - unlike Size, they exclude batches currently in + /// progress + /// + Task GetAdjunctQueue(int customer, CancellationToken cancellationToken); + /// /// Increment specified queue by specified amount. /// If queue doesn't exist it will be created with Size = incrementAmount diff --git a/src/protagonist/DLCS.Repository/Migrations/DlcsContextModelSnapshot.cs b/src/protagonist/DLCS.Repository/Migrations/DlcsContextModelSnapshot.cs index 86e07c964..feeccdef7 100644 --- a/src/protagonist/DLCS.Repository/Migrations/DlcsContextModelSnapshot.cs +++ b/src/protagonist/DLCS.Repository/Migrations/DlcsContextModelSnapshot.cs @@ -1276,12 +1276,14 @@ protected override void BuildModel(ModelBuilder modelBuilder) .OnDelete(DeleteBehavior.Cascade) .IsRequired(); - b.HasOne("DLCS.Model.Assets.Adjunct", null) + b.HasOne("DLCS.Model.Assets.Adjunct", "Adjunct") .WithMany("AdjunctBatchAdjuncts") .HasForeignKey("AdjunctId", "AssetId") .OnDelete(DeleteBehavior.Cascade) .IsRequired(); + b.Navigation("Adjunct"); + b.Navigation("Batch"); }); diff --git a/src/protagonist/DLCS.Repository/Processing/CustomerQueueRepository.cs b/src/protagonist/DLCS.Repository/Processing/CustomerQueueRepository.cs index d91d85a78..75d89aa5a 100644 --- a/src/protagonist/DLCS.Repository/Processing/CustomerQueueRepository.cs +++ b/src/protagonist/DLCS.Repository/Processing/CustomerQueueRepository.cs @@ -3,6 +3,7 @@ using System.Threading; using System.Threading.Tasks; using DLCS.Model.Processing; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; namespace DLCS.Repository.Processing; @@ -38,6 +39,31 @@ public CustomerQueueRepository(DlcsContext dlcsContext, ILogger GetAdjunctQueue(int customer, CancellationToken cancellationToken) + { + try + { + // BatchesWaiting/AdjunctsWaiting only count batches that haven't started processing yet (nothing + // Completed or Errored) - this excludes batches currently being worked on, which still count towards + // Size. Raw SQL rather than EF, as above - a single scan of AdjunctBatches via the subquery, rather + // than the two correlated-subquery scans EF would generate for BatchesWaiting/AdjunctsWaiting. + const string sql = @"SELECT +q.""Customer"", q.""Size"", b.""BatchesWaiting"", b.""AdjunctsWaiting"" FROM ""Queues"" q, + (SELECT COUNT(""Id"") AS ""BatchesWaiting"", COALESCE(SUM(""Count""), 0) AS ""AdjunctsWaiting"" + FROM ""AdjunctBatches"" WHERE ""Customer"" = @customer AND ""Finished"" IS NULL AND ""Completed"" = 0 + AND ""Errors"" = 0) b + WHERE q.""Customer"" = @customer AND ""Name"" = @name +"; + return await this.QueryFirstOrDefaultAsync(sql, + new { customer, name = QueueNames.Adjunct }); + } + catch (Exception ex) + { + logger.LogError(ex, "Error getting adjunct queue counts for customer {Customer}", customer); + return null; + } + } + public async Task IncrementSize(int customer, string name, int incrementAmount = 1, CancellationToken cancellationToken = default) { @@ -73,4 +99,4 @@ await DlcsContext.Queues.AddAsync( amount); } } -} \ No newline at end of file +} diff --git a/src/protagonist/Engine.Tests/Ingest/Handlers/IngestHandlerTests.cs b/src/protagonist/Engine.Tests/Ingest/Handlers/IngestHandlerTests.cs index 8498bfcd5..631eebb67 100644 --- a/src/protagonist/Engine.Tests/Ingest/Handlers/IngestHandlerTests.cs +++ b/src/protagonist/Engine.Tests/Ingest/Handlers/IngestHandlerTests.cs @@ -89,4 +89,32 @@ public async Task HandleMessage_ReturnsFalse_IfSuccessOrQueued(IngestResultStatu .MustHaveHappened(); success.Should().BeTrue(); } + + [Fact] + public async Task HandleMessage_DecrementsAdjunctQueue_NotDefault_ForAdjunctIngestMessage() + { + // Arrange + var body = new JsonObject + { + ["created"] = "1985-10-26T09:00:00" + }; + var queueMessage = new QueueMessage + { + Body = body, QueueName = "protagonist-adjunct", + MessageAttributes = new() { { SqsQueueUtilities.Constants.MessageAttributeNames.IngestType, IngestAdjunctRequest.IngestType } } + }; + A.CallTo(() => adjunctIngester.Ingest(A._, A._)) + .Returns(new IngestResult(new AssetId(1, 2, "fake"), IngestResultStatus.Success)); + + // Act + var success = await sut.HandleMessage(queueMessage, CancellationToken.None); + + // Assert + A.CallTo(() => adjunctIngester.Ingest(A._, A._)).MustHaveHappened(); + A.CallTo(() => customerQueueRepository.DecrementSize(A._, QueueNames.Adjunct, A._, A._)) + .MustHaveHappened(); + A.CallTo(() => customerQueueRepository.DecrementSize(A._, QueueNames.Default, A._, A._)) + .MustNotHaveHappened(); + success.Should().BeTrue(); + } } diff --git a/src/protagonist/Engine/Ingest/IngestHandler.cs b/src/protagonist/Engine/Ingest/IngestHandler.cs index 4fbc85c06..0ce74f646 100644 --- a/src/protagonist/Engine/Ingest/IngestHandler.cs +++ b/src/protagonist/Engine/Ingest/IngestHandler.cs @@ -46,7 +46,7 @@ private async Task HandleIngest(QueueMessage message, logger.LogDebug("Message {MessageId} handled with result {IngestResult}", message.MessageId, ingestResult.Status); - await UpdateCustomerQueue(message, ingestResult, cancellationToken); + await UpdateCustomerQueue(ingestResult, GetQueueName(message), cancellationToken); } // return true so that the message is deleted from the queue in all instances. @@ -56,10 +56,16 @@ private async Task HandleIngest(QueueMessage message, return true; } - private async Task UpdateCustomerQueue(QueueMessage message, - IngestResult ingestResult, CancellationToken cancellationToken) + private static string GetQueueName(QueueMessage message) => message switch + { + _ when typeof(T) == typeof(IngestAdjunctRequest) => QueueNames.Adjunct, + _ when message.QueueName.Contains("priority", StringComparison.OrdinalIgnoreCase) => QueueNames.Priority, + _ => QueueNames.Default + }; + + private async Task UpdateCustomerQueue( + IngestResult ingestResult, string queue, CancellationToken cancellationToken) { - var queue = message.QueueName.Contains("priority", StringComparison.OrdinalIgnoreCase) ? QueueNames.Priority : QueueNames.Default; var customer = 0; try { diff --git a/src/protagonist/Test.Helpers/Integration/DatabaseTestDataPopulation.cs b/src/protagonist/Test.Helpers/Integration/DatabaseTestDataPopulation.cs index 3f7a64ccf..0778e133a 100644 --- a/src/protagonist/Test.Helpers/Integration/DatabaseTestDataPopulation.cs +++ b/src/protagonist/Test.Helpers/Integration/DatabaseTestDataPopulation.cs @@ -307,11 +307,16 @@ public static ValueTask> AddTestBatchAsset(this DbSet batchAssets.AddAsync(new BatchAsset { AssetId = assetId, BatchId = batchId }); public static ValueTask> AddTestAdjunctBatch(this DbSet adjunctBatches, - int id, int customer = 99, int count = 1, int completed = 0, int errors = 0, + int? id = null, int customer = 99, int count = 1, int completed = 0, int errors = 0, DateTime? submitted = null, DateTime? finished = null) - => adjunctBatches.AddAsync(new AdjunctBatch + { + var batch = new AdjunctBatch { - Id = id, Customer = customer, Submitted = submitted ?? DateTime.UtcNow, + Customer = customer, Submitted = submitted ?? DateTime.UtcNow, Count = count, Completed = completed, Errors = errors, Finished = finished - }); + }; + if (id.HasValue) batch.Id = id.Value; + + return adjunctBatches.AddAsync(batch); + } } diff --git a/src/protagonist/Test.Helpers/Integration/DlcsDatabaseFixture.cs b/src/protagonist/Test.Helpers/Integration/DlcsDatabaseFixture.cs index d3c077b04..a86cc7df0 100644 --- a/src/protagonist/Test.Helpers/Integration/DlcsDatabaseFixture.cs +++ b/src/protagonist/Test.Helpers/Integration/DlcsDatabaseFixture.cs @@ -41,6 +41,8 @@ public void CleanUp() DbContext.Database.ExecuteSqlRaw("DELETE FROM \"AuthTokens\""); DbContext.Database.ExecuteSqlRaw("DELETE FROM \"Batches\""); DbContext.Database.ExecuteSqlRaw("DELETE FROM \"Queues\""); + DbContext.Database.ExecuteSqlRaw("DELETE FROM \"AdjunctBatchAdjuncts\""); + DbContext.Database.ExecuteSqlRaw("DELETE FROM \"AdjunctBatches\""); DbContext.Database.ExecuteSqlRaw("DELETE FROM \"EntityCounters\" WHERE \"Type\" = 'space' AND \"Customer\" != 99"); DbContext.Database.ExecuteSqlRaw("DELETE FROM \"EntityCounters\" WHERE \"Type\" = 'space-images' AND \"Customer\" != 99"); DbContext.Database.ExecuteSqlRaw("DELETE FROM \"EntityCounters\" WHERE \"Type\" = 'customer-images' AND \"Scope\" != '99'");