Consumer and service for update status sms (#409)

* Consumer and service for update status sms

* Fixed codesmells

* re-wrote get tests

* fixed code smell

* Update src/Altinn.Notifications.Core/Models/Notification/EmailSendOperationResult.cs

Co-authored-by: Stephanie Buadu <47737608+acn-sbuad@users.noreply.github.com>

* Removed nullable from SmsSendOperationResult

* Update test/Altinn.Notifications.IntegrationTests/Notifications/EmailNotificationsController/GetTests.cs

* Add test for invalid SmsSendOperationResult

* Adjust test for failed parsing

---------

Co-authored-by: acn-sbuad <stephanie.buadu@avanade.com>
Co-authored-by: Stephanie Buadu <47737608+acn-sbuad@users.noreply.github.com>

src/Altinn.Notifications.Core/Models/Notification/SendOperationResult.cs → src/Altinn.Notifications.Core/Models/Notification/EmailSendOperationResult.cs

@@ -5,9 +5,9 @@
 namespace Altinn.Notifications.Core.Models.Notification;
 
 /// <summary>
-/// A class representing a send operation update object
+/// A class representing an email send operation result object
 /// </summary>                              
-public class SendOperationResult
+public class EmailSendOperationResult
 {
     /// <summary>
     /// The notification id
@@ -25,29 +25,29 @@ public class SendOperationResult
     public EmailNotificationResultType? SendResult { get; set; }
 
     /// <summary>
-    /// Json serializes the <see cref="SendOperationResult"/>
+    /// Json serializes the <see cref="EmailSendOperationResult"/>
     /// </summary>
     public string Serialize()
     {
         return JsonSerializer.Serialize(this, JsonSerializerOptionsProvider.Options);
     }
 
     /// <summary>
-    /// Deserialize a json string into the <see cref="SendOperationResult"/>
+    /// Deserialize a json string into the <see cref="EmailSendOperationResult"/>
     /// </summary>
-    public static SendOperationResult? Deserialize(string serializedString)
+    public static EmailSendOperationResult? Deserialize(string serializedString)
     {
-        return JsonSerializer.Deserialize<SendOperationResult>(
+        return JsonSerializer.Deserialize<EmailSendOperationResult>(
             serializedString, JsonSerializerOptionsProvider.Options);
     }
 
     /// <summary>
-    /// Try to parse a json string into a<see cref="SendOperationResult"/>
+    /// Try to parse a json string into a<see cref="EmailSendOperationResult"/>
     /// </summary>
-    public static bool TryParse(string input, out SendOperationResult value)
+    public static bool TryParse(string input, out EmailSendOperationResult value)
     {
-        SendOperationResult? parsedOutput;
-        value = new SendOperationResult();
+        EmailSendOperationResult? parsedOutput;
+        value = new EmailSendOperationResult();
 
         if (string.IsNullOrEmpty(input))
         {

src/Altinn.Notifications.Core/Models/Notification/SmsSendOperationResult.cs

@@ -0,0 +1,71 @@
+using System.Text.Json;
+
+using Altinn.Notifications.Core.Enums;
+
+namespace Altinn.Notifications.Core.Models.Notification;
+
+/// <summary>
+/// A class representing a sms send operation update object
+/// </summary>    
+public class SmsSendOperationResult
+{
+    /// <summary>
+    /// The notification id
+    /// </summary>
+    public Guid NotificationId { get; set; }
+
+    /// <summary>
+    /// The reference to the delivery in sms gateway
+    /// </summary>
+    public string GatewayReference { get; set; } = string.Empty;
+
+    /// <summary>
+    /// The sms send result
+    /// </summary>
+    public SmsNotificationResultType SendResult { get; set; }
+
+    /// <summary>
+    /// Json serializes the <see cref="SmsSendOperationResult"/>
+    /// </summary>
+    public string Serialize()
+    {
+        return JsonSerializer.Serialize(this, JsonSerializerOptionsProvider.Options);
+    }
+
+    /// <summary>
+    /// Deserialize a json string into the <see cref="SmsSendOperationResult"/>
+    /// </summary>
+    public static SmsSendOperationResult? Deserialize(string serializedString)
+    {
+        return JsonSerializer.Deserialize<SmsSendOperationResult>(
+            serializedString, JsonSerializerOptionsProvider.Options);
+    }
+
+    /// <summary>
+    /// Try to parse a json string into a<see cref="SmsSendOperationResult"/>
+    /// </summary>
+    public static bool TryParse(string input, out SmsSendOperationResult value)
+    {
+        SmsSendOperationResult? parsedOutput;
+        value = new SmsSendOperationResult();
+
+        if (string.IsNullOrEmpty(input))
+        {
+            return false;
+        }
+
+        try
+        {
+            parsedOutput = Deserialize(input!);
+
+            value = parsedOutput!;
+            return value.NotificationId != Guid.Empty;
+        }
+        catch
+        {
+            // try parse, we simply return false if fails
+        }
+
+        return false;
+    }
+}

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

@@ -70,7 +70,7 @@ public async Task SendNotifications()
     }
 
     /// <inheritdoc/>
-    public async Task UpdateSendStatus(SendOperationResult sendOperationResult)
+    public async Task UpdateSendStatus(EmailSendOperationResult sendOperationResult)
     {
         // set to new to allow new iteration of regular proceessing if transient error
         if (sendOperationResult.SendResult == EmailNotificationResultType.Failed_TransientError)

src/Altinn.Notifications.Core/Services/Interfaces/IEmailNotificationService.cs

@@ -21,5 +21,5 @@ public interface IEmailNotificationService
     /// <summary>
     /// Update send status for a notification
     /// </summary>
-    public Task UpdateSendStatus(SendOperationResult sendOperationResult);
+    public Task UpdateSendStatus(EmailSendOperationResult sendOperationResult);
 }

src/Altinn.Notifications.Core/Services/Interfaces/ISmsNotificationService.cs

@@ -1,4 +1,5 @@
 using Altinn.Notifications.Core.Models;
+using Altinn.Notifications.Core.Models.Notification;
 
 namespace Altinn.Notifications.Core.Services.Interfaces;
 
@@ -16,4 +17,9 @@ public interface ISmsNotificationService
     /// Starts the process of sending all ready sms notifications
     /// </summary>
     public Task SendNotifications();
+
+    /// <summary>
+    /// Update send status for an sms notification
+    /// </summary>
+    public Task UpdateSendStatus(SmsSendOperationResult sendOperationResult);
 }

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

@@ -69,6 +69,12 @@ public async Task SendNotifications()
         }
     }
 
+    /// <inheritdoc/>
+    public async Task UpdateSendStatus(SmsSendOperationResult sendOperationResult)
+    {
+        await _repository.UpdateSendStatus(sendOperationResult.NotificationId, sendOperationResult.SendResult, sendOperationResult.GatewayReference);
+    }
+
     private async Task CreateNotificationForRecipient(Guid orderId, DateTime requestedSendTime, string recipientId, string recipientNumber, SmsNotificationResultType type)
     {
         var smsNotification = new SmsNotification()

src/Altinn.Notifications.Integrations/Configuration/KafkaSettings.cs

@@ -45,6 +45,11 @@ public class KafkaSettings
     /// </summary>
     public string EmailStatusUpdatedTopicName { get; set; } = string.Empty;
 
+    /// <summary>
+    /// The name of the sms status updated topic
+    /// </summary>
+    public string SmsStatusUpdatedTopicName { get; set; } = string.Empty;
+
     /// <summary>
     /// The name of the platform service update topic 
     /// </summary>

src/Altinn.Notifications.Integrations/Extensions/ServiceCollectionExtensions.cs

@@ -30,6 +30,7 @@ public static void AddKafkaServices(this IServiceCollection services, IConfigura
         .AddHostedService<PastDueOrdersConsumer>()
         .AddHostedService<PastDueOrdersRetryConsumer>()
         .AddHostedService<EmailStatusConsumer>()
+        .AddHostedService<SmsStatusConsumer>()
         .AddHostedService<AltinnServiceUpdateConsumer>()
         .Configure<KafkaSettings>(config.GetSection(nameof(KafkaSettings)));
     }

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

@@ -42,7 +42,7 @@ protected override Task ExecuteAsync(CancellationToken stoppingToken)
 
     private async Task ProcessStatus(string message)
     {
-        bool succeeded = SendOperationResult.TryParse(message, out SendOperationResult result);
+        bool succeeded = EmailSendOperationResult.TryParse(message, out EmailSendOperationResult result);
 
         if (!succeeded)
         {

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

@@ -0,0 +1,59 @@
+using Altinn.Notifications.Core.Integrations;
+using Altinn.Notifications.Core.Models.Notification;
+using Altinn.Notifications.Core.Services.Interfaces;
+using Altinn.Notifications.Integrations.Configuration;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Altinn.Notifications.Integrations.Kafka.Consumers;
+
+/// <summary>
+/// Kafka consumer class for status messages about sms notifications
+/// </summary>
+public class SmsStatusConsumer : KafkaConsumerBase<SmsStatusConsumer>
+{
+    private readonly ISmsNotificationService _smsNotificationsService;
+    private readonly IKafkaProducer _producer;
+    private readonly string _retryTopicName;
+    private readonly ILogger<SmsStatusConsumer> _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="SmsStatusConsumer"/> class.
+    /// </summary>
+    public SmsStatusConsumer(
+        ISmsNotificationService smsNotificationsService,
+        IKafkaProducer producer,
+        IOptions<KafkaSettings> settings,
+        ILogger<SmsStatusConsumer> logger)
+        : base(settings, logger, settings.Value.SmsStatusUpdatedTopicName)
+    {
+        _smsNotificationsService = smsNotificationsService;
+        _producer = producer;
+        _retryTopicName = settings.Value.SmsStatusUpdatedTopicName;
+        _logger = logger;
+    }
+
+    /// <inheritdoc/>
+    protected override Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        return Task.Run(() => ConsumeMessage(ProcessStatus, RetryStatus, stoppingToken), stoppingToken);
+    }
+
+    private async Task ProcessStatus(string message)
+    {
+        bool succeeded = SmsSendOperationResult.TryParse(message, out SmsSendOperationResult result);
+
+        if (!succeeded)
+        {
+            _logger.LogError("// SmsStatusConsumer // ProcessStatus // Deserialization of message failed. {Message}", message);
+            return;
+        }
+
+        await _smsNotificationsService.UpdateSendStatus(result);
+    }
+
+    private async Task RetryStatus(string message)
+    {
+        await _producer.ProduceAsync(_retryTopicName, message!);
+    }
+}

src/Altinn.Notifications/appsettings.json

@@ -25,6 +25,7 @@
     "EmailQueueTopicName": "altinn.notifications.email.queue",
     "EmailStatusUpdatedTopicName": "altinn.notifications.email.status.updated",
     "SmsQueueTopicName": "altinn.notifications.sms.queue",
+    "SmsStatusUpdatedTopicName": "altinn.notifications.sms.status.updated",
     "HealthCheckTopic": "altinn.notifications.health.check",
     "AltinnServiceUpdateTopicName": "altinn.platform.service.updated",
     "Admin": {
@@ -35,6 +36,7 @@
         "altinn.notifications.email.queue",
         "altinn.notifications.email.status.updated",
         "altinn.notifications.sms.queue",
+        "altinn.notifications.sms.status.updated",
         "altinn.platform.service.updated"
 
       ]

test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/EmailStatusConsumerTests.cs

@@ -30,7 +30,7 @@ public async Task RunTask_ConfirmExpectedSideEffects()
 
         (_, EmailNotification notification) = await PostgreUtil.PopulateDBWithOrderAndEmailNotification(_sendersRef);
 
-        SendOperationResult sendOperationResult = new()
+        EmailSendOperationResult sendOperationResult = new()
         {
             NotificationId = notification.Id,
             OperationId = Guid.NewGuid().ToString(),

test/Altinn.Notifications.IntegrationTests/Notifications.Integrations/TestingConsumers/SmsStatusConsumerTests.cs

@@ -0,0 +1,100 @@
+using Altinn.Notifications.Core.Enums;
+using Altinn.Notifications.Core.Models.Notification;
+using Altinn.Notifications.Integrations.Kafka.Consumers;
+using Altinn.Notifications.IntegrationTests.Utils;
+using Microsoft.Extensions.Hosting;
+
+using Xunit;
+
+namespace Altinn.Notifications.IntegrationTests.Notifications.Integrations.TestingConsumers;
+
+public class SmsStatusConsumerTests : IAsyncLifetime
+{
+    private readonly string _statusUpdatedTopicName = Guid.NewGuid().ToString();
+    private readonly string _sendersRef = $"ref-{Guid.NewGuid()}";
+
+    [Fact]
+    public async Task RunTask_ConfirmExpectedSideEffects()
+    {
+        // Arrange
+        Dictionary<string, string> vars = new()
+        {
+            { "KafkaSettings__SmsStatusUpdatedTopicName", _statusUpdatedTopicName },
+            { "KafkaSettings__Admin__TopicList", $"[\"{_statusUpdatedTopicName}\"]" }
+        };
+
+        using SmsStatusConsumer consumerService = (SmsStatusConsumer)ServiceUtil
+                                            .GetServices(new List<Type>() { typeof(IHostedService) }, vars)
+                                            .First(s => s.GetType() == typeof(SmsStatusConsumer))!;
+
+        (_, SmsNotification notification) = await PostgreUtil.PopulateDBWithOrderAndSmsNotification(_sendersRef);
+
+        SmsSendOperationResult sendOperationResult = new()
+        {
+            NotificationId = notification.Id,
+            SendResult = SmsNotificationResultType.Accepted,
+            GatewayReference = Guid.NewGuid().ToString()
+        };
+
+        await KafkaUtil.PublishMessageOnTopic(_statusUpdatedTopicName, sendOperationResult.Serialize());
+
+        // Act
+        await consumerService.StartAsync(CancellationToken.None);
+        await Task.Delay(10000);
+        await consumerService.StopAsync(CancellationToken.None);
+
+        // Assert
+        string smsNotificationStatus = await SelectSmsNotificationStatus(notification.Id);
+        Assert.Equal(SmsNotificationResultType.Accepted.ToString(), smsNotificationStatus);
+    }
+
+    [Fact]
+    public async Task RunTask_ParseSmsSendOperationResult_StatusNotUpdated()
+    {
+        // Arrange
+        Dictionary<string, string> vars = new()
+        {
+            { "KafkaSettings__SmsStatusUpdatedTopicName", _statusUpdatedTopicName },
+            { "KafkaSettings__Admin__TopicList", $"[\"{_statusUpdatedTopicName}\"]" }
+        };
+
+        using SmsStatusConsumer consumerService = (SmsStatusConsumer)ServiceUtil
+                                            .GetServices(new List<Type>() { typeof(IHostedService) }, vars)
+                                            .First(s => s.GetType() == typeof(SmsStatusConsumer))!;
+
+        (_, SmsNotification notification) = await PostgreUtil.PopulateDBWithOrderAndSmsNotification(_sendersRef);
+
+        await KafkaUtil.PublishMessageOnTopic(_statusUpdatedTopicName, string.Empty);
+
+        // Act
+        await consumerService.StartAsync(CancellationToken.None);
+        await Task.Delay(10000);
+        await consumerService.StopAsync(CancellationToken.None);
+
+        // Assert
+        string smsNotificationStatus = await SelectSmsNotificationStatus(notification.Id);
+        Assert.Equal(SmsNotificationResultType.New.ToString(), smsNotificationStatus);
+    }
+
+    public Task InitializeAsync()
+    {
+        return Task.CompletedTask;
+    }
+
+    public async Task DisposeAsync()
+    {
+        await Dispose(true);
+    }
+
+    protected virtual async Task Dispose(bool disposing)
+    {
+        await PostgreUtil.DeleteOrderFromDb(_sendersRef);
+        await KafkaUtil.DeleteTopicAsync(_statusUpdatedTopicName);
+    }
+
+    private static async Task<string> SelectSmsNotificationStatus(Guid notificationId)
+    {
+        string sql = $"select result from notifications.smsnotifications where alternateid = '{notificationId}'";
+        return await PostgreUtil.RunSqlReturnOutput<string>(sql);
+    }
+}