Added condition check to order processing

dbsetup.sh

@@ -1,6 +1,10 @@
  #!/bin/bash
 export PGPASSWORD=Password
 
+# alter max connections
+psql -h localhost -p 5432 -U platform_notifications_admin -d notificationsdb  \
+-c "ALTER SYSTEM SET max_connections TO '200';"
+
 # set up platform_notifications role
 psql -h localhost -p 5432 -U platform_notifications_admin -d notificationsdb \
 -c "DO \$\$

src/Altinn.Notifications.Core/Enums/OrderProcessingStatus.cs

@@ -8,6 +8,7 @@ public enum OrderProcessingStatus
 {
     Registered,
     Processing,
-    Completed
+    Completed,
+    SendConditionNotMet
 }
 #pragma warning restore CS1591 // Missing XML comment for publicly visible type or member

src/Altinn.Notifications.Core/Exceptions/OrderProcessingException.cs

@@ -0,0 +1,33 @@
+namespace Altinn.Notifications.Core.Exceptions;
+
+/// <summary>
+/// Represents errors that occur during order processing operations.
+/// </summary>
+public class OrderProcessingException : Exception
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="OrderProcessingException"/> class.
+    /// </summary>
+    public OrderProcessingException() : base()
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="OrderProcessingException"/> class
+    /// with a specified error message.
+    /// </summary>
+    /// <param name="message">The message that describes the error.</param>
+    public OrderProcessingException(string message) : base(message)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="OrderProcessingException"/> class
+    /// with a specified error message and a reference to the inner exception that is the cause of this exception.
+    /// </summary>
+    /// <param name="message">The error message that explains the reason for the exception.</param>
+    /// <param name="inner">The exception that is the cause of the current exception, or a null reference if no inner exception is specified.</param>
+    public OrderProcessingException(string message, Exception inner) : base(message, inner)
+    {
+    }
+}

src/Altinn.Notifications.Core/Services/GetOrderService.cs

@@ -17,6 +17,7 @@ public class GetOrderService : IGetOrderService
             { OrderProcessingStatus.Registered, "Order has been registered and is awaiting requested send time before processing." },
             { OrderProcessingStatus.Processing, "Order processing is ongoing. Notifications are being generated." },
             { OrderProcessingStatus.Completed, "Order processing is completed. All notifications have been generated." },
+            { OrderProcessingStatus.SendConditionNotMet, "Order processing was stopped due to send condition not being met."}
         };
 
     /// <summary>

src/Altinn.Notifications.Core/Services/OrderProcessingService.cs

@@ -2,11 +2,13 @@
 
 using Altinn.Notifications.Core.Configuration;
 using Altinn.Notifications.Core.Enums;
+using Altinn.Notifications.Core.Exceptions;
 using Altinn.Notifications.Core.Integrations;
 using Altinn.Notifications.Core.Models.Orders;
 using Altinn.Notifications.Core.Persistence;
 using Altinn.Notifications.Core.Services.Interfaces;
 
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
 namespace Altinn.Notifications.Core.Services;
@@ -19,8 +21,10 @@ public class OrderProcessingService : IOrderProcessingService
     private readonly IOrderRepository _orderRepository;
     private readonly IEmailOrderProcessingService _emailProcessingService;
     private readonly ISmsOrderProcessingService _smsProcessingService;
+    private readonly IConditionClient _conditionClient;
     private readonly IKafkaProducer _producer;
     private readonly string _pastDueOrdersTopic;
+    private readonly ILogger<OrderProcessingService> _logger;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="OrderProcessingService"/> class.
@@ -29,14 +33,18 @@ public OrderProcessingService(
         IOrderRepository orderRepository,
         IEmailOrderProcessingService emailProcessingService,
         ISmsOrderProcessingService smsProcessingService,
+        IConditionClient conditionClient,
         IKafkaProducer producer,
-        IOptions<KafkaSettings> kafkaSettings)
+        IOptions<KafkaSettings> kafkaSettings,
+        ILogger<OrderProcessingService> logger)
     {
         _orderRepository = orderRepository;
         _emailProcessingService = emailProcessingService;
         _smsProcessingService = smsProcessingService;
+        _conditionClient = conditionClient;
         _producer = producer;
         _pastDueOrdersTopic = kafkaSettings.Value.PastDueOrdersTopicName;
+        _logger = logger;
     }
 
     /// <inheritdoc/>
@@ -65,6 +73,12 @@ public async Task StartProcessingPastDueOrders()
     /// <inheritdoc/>
     public async Task ProcessOrder(NotificationOrder order)
     {
+        if (!await IsSendConditionMet(order, isRetry: false))
+        {
+            await _orderRepository.SetProcessingStatus(order.Id, OrderProcessingStatus.SendConditionNotMet);
+            return;
+        }
+
         NotificationChannel ch = order.NotificationChannel;
 
         switch (ch)
@@ -83,6 +97,12 @@ public async Task ProcessOrder(NotificationOrder order)
     /// <inheritdoc/>
     public async Task ProcessOrderRetry(NotificationOrder order)
     {
+        if (!await IsSendConditionMet(order, isRetry: true))
+        {
+            await _orderRepository.SetProcessingStatus(order.Id, OrderProcessingStatus.SendConditionNotMet);
+            return;
+        }
+
         NotificationChannel ch = order.NotificationChannel;
 
         switch (ch)
@@ -97,4 +117,33 @@ public async Task ProcessOrderRetry(NotificationOrder order)
 
         await _orderRepository.SetProcessingStatus(order.Id, OrderProcessingStatus.Completed);
     }
+
+    private async Task<bool> IsSendConditionMet(NotificationOrder order, bool isRetry)
+    {
+        if (order.ConditionEndpoint == null)
+        {
+            return true;
+        }
+
+        var conditionCheckResult = await _conditionClient.CheckSendCondition(order.ConditionEndpoint);
+
+        return conditionCheckResult.Match(
+          successResult =>
+          {
+              return successResult;
+          },
+          errorResult =>
+          {
+              if (!isRetry)
+              {
+                  // always send to retry on first error.
+                  throw new OrderProcessingException($"// OrderProcessingService // IsSendConditionMet // Condition check for order with ID '{order.Id}' failed with HTTP status code '{errorResult.StatusCode}' at endpoint '{order.ConditionEndpoint}'");
+              }
+
+              // notifications should always be created and sent if the condition check is not successful
+              // log error. 
+              _logger.LogInformation("// OrderProcessingService // IsSendConditionMet // Condition check for order with ID '{order.Id}' failed on retry. Processing regardless.", order.Id);
+              return true;
+          });
+    }
 }

src/Altinn.Notifications.Integrations/Kafka/Consumers/PastDueOrdersRetryConsumer.cs

@@ -13,17 +13,22 @@ namespace Altinn.Notifications.Integrations.Kafka.Consumers;
 public class PastDueOrdersRetryConsumer : KafkaConsumerBase<PastDueOrdersRetryConsumer>
 {
     private readonly IOrderProcessingService _orderProcessingService;
+    private readonly IDateTimeService _dateTime;
+
+    private readonly int _processingDelay = 60000;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="PastDueOrdersRetryConsumer"/> class.
     /// </summary>
     public PastDueOrdersRetryConsumer(
         IOrderProcessingService orderProcessingService,
+        IDateTimeService dateTimeService,
         IOptions<KafkaSettings> settings,
         ILogger<PastDueOrdersRetryConsumer> logger)
         : base(settings, logger, settings.Value.PastDueOrdersRetryTopicName)
     {
         _orderProcessingService = orderProcessingService;
+        _dateTime = dateTimeService;
     }
 
     /// <inheritdoc/>
@@ -41,6 +46,14 @@ private async Task ProcessOrder(string message)
             return;
         }
 
+        // adding a delay relative to send time to allow transient faults to be resolved
+        int diff = (int)(_dateTime.UtcNow() - order.RequestedSendTime).TotalMilliseconds;
+
+        if (diff < _processingDelay)
+        {
+            await Task.Delay(_processingDelay - diff);
+        }
+
         await _orderProcessingService.ProcessOrderRetry(order!);
     }
 

test/Altinn.Notifications.Tests/Notifications.Core/TestingServices/GetOrderServiceTests.cs

@@ -135,6 +135,7 @@ await result.Match<Task>(
     [InlineData(OrderProcessingStatus.Registered, "Order has been registered and is awaiting requested send time before processing.")]
     [InlineData(OrderProcessingStatus.Processing, "Order processing is ongoing. Notifications are being generated.")]
     [InlineData(OrderProcessingStatus.Completed, "Order processing is completed. All notifications have been generated.")]
+    [InlineData(OrderProcessingStatus.SendConditionNotMet, "Order processing was stopped due to send condition not being met.")]
     public void GetStatusDescription_ExpectedDescription(OrderProcessingStatus status, string expected)
     {
         string actual = GetOrderService.GetStatusDescription(status);

test/Altinn.Notifications.Tests/Notifications.Core/TestingServices/OrderProcessingServiceTests.cs

@@ -9,6 +9,9 @@
 using Altinn.Notifications.Core.Services;
 using Altinn.Notifications.Core.Services.Interfaces;
 
+using Castle.Core.Logging;
+
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
 using Moq;
@@ -179,7 +182,8 @@ private static OrderProcessingService GetTestService(
         IOrderRepository? repo = null,
         IEmailOrderProcessingService? emailMock = null,
         ISmsOrderProcessingService? smsMock = null,
-        IKafkaProducer? producer = null)
+        IKafkaProducer? producer = null,
+        IConditionClient? conditionClient = null)
     {
         if (repo == null)
         {
@@ -205,8 +209,14 @@ private static OrderProcessingService GetTestService(
             producer = producerMock.Object;
         }
 
+        if (conditionClient == null)
+        {
+            var conditionClientMock = new Mock<IConditionClient>();
+            conditionClient = conditionClientMock.Object;
+        }
+
         var kafkaSettings = new Altinn.Notifications.Core.Configuration.KafkaSettings() { PastDueOrdersTopicName = _pastDueTopicName };
 
-        return new OrderProcessingService(repo, emailMock, smsMock, producer, Options.Create(kafkaSettings));
+        return new OrderProcessingService(repo, emailMock, smsMock, conditionClient, producer, Options.Create(kafkaSettings), new LoggerFactory().CreateLogger<OrderProcessingService>());
     }
 }