Feature/duplicate confirm handling (#331)

* Duplicate handling on confirm Download

* Add start script to run dev

* Format

* Resolve conversation + cleanup

* rethrow error message after deletion

---------

Co-authored-by: Hammerbeck <andreas.hammerbeck@digdir.no>

src/Altinn.Broker.API/Controllers/FileController.cs

@@ -10,6 +10,7 @@
 using Altinn.Broker.Core.Domain;
 using Altinn.Broker.Core.Domain.Enums;
 using Altinn.Broker.Core.Models;
+using Altinn.Broker.Core.Repositories;
 using Altinn.Broker.Enums;
 using Altinn.Broker.Helpers;
 using Altinn.Broker.Mappers;
@@ -20,6 +21,8 @@
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 
+using OneOf;
+
 namespace Altinn.Broker.Controllers
 {
     [ApiController]
@@ -28,10 +31,12 @@ namespace Altinn.Broker.Controllers
     public class FileController : Controller
     {
         private readonly ILogger<FileController> _logger;
+        private readonly IIdempotencyEventRepository _idempotencyEventRepository;
 
-        public FileController(ILogger<FileController> logger)
+        public FileController(ILogger<FileController> logger, IIdempotencyEventRepository idempotencyEventRepository)
         {
             _logger = logger;
+            _idempotencyEventRepository = idempotencyEventRepository;
         }
 
         /// <summary>
@@ -250,11 +255,14 @@ public async Task<ActionResult> ConfirmDownload(
         {
             LogContextHelpers.EnrichLogsWithToken(token);
             _logger.LogInformation("Confirming download for file {fileId}", fileId.ToString());
-            var commandResult = await handler.Process(new ConfirmDownloadCommandRequest()
+            var requestData = new ConfirmDownloadCommandRequest()
             {
                 FileId = fileId,
                 Token = token
-            }, cancellationToken);
+            };
+            var processingFunction = new Func<Task<OneOf<Task, Error>>>(() => handler.Process(requestData, cancellationToken));
+            var uniqueString = $"confirmDownload_{fileId}_{token.Consumer}";
+            var commandResult = await IdempotencyEventHelper.ProcessEvent(uniqueString, processingFunction, _idempotencyEventRepository, cancellationToken);
             return commandResult.Match(
                 (_) => Ok(null),
                 Problem

src/Altinn.Broker.API/Controllers/MalwareScanResultsController.cs

@@ -11,6 +11,8 @@
 
 using Newtonsoft.Json;
 
+using OneOf;
+
 namespace Altinn.Broker.Webhooks.Controllers
 {
     [ApiController]
@@ -19,13 +21,13 @@ public class MalwareScanResultsController : Controller
     {
         private readonly IFileRepository _fileRepository;
         private readonly IFileStatusRepository _fileStatusRepository;
-        private readonly IWebhookEventRepository _webhookEventRepository;
+        private readonly IIdempotencyEventRepository _idempotencyEventRepository;
 
-        public MalwareScanResultsController(IFileRepository fileRepository, IFileStatusRepository fileStatusRepository, IWebhookEventRepository webhookEventRepository)
+        public MalwareScanResultsController(IFileRepository fileRepository, IFileStatusRepository fileStatusRepository, IIdempotencyEventRepository idempotencyEventRepository)
         {
             _fileRepository = fileRepository;
             _fileStatusRepository = fileStatusRepository;
-            _webhookEventRepository = webhookEventRepository;
+            _idempotencyEventRepository = idempotencyEventRepository;
         }
 
         [HttpPost]
@@ -51,7 +53,8 @@ public async Task<ActionResult> ProcessMalwareScanResult([FromServices] MalwareS
                 {
                     string jsonString = eventGridEvent.Data.ToString();
                     ScanResultData result = JsonConvert.DeserializeObject<ScanResultData>(jsonString);
-                    var commandResult = await WebhookEventHelper.ProcessMalwareEvent(result, handler, _webhookEventRepository, cancellationToken);
+                    var processFunction = new Func<Task<OneOf<Task, Error>>>(() => handler.Process(result, cancellationToken));
+                    var commandResult = await IdempotencyEventHelper.ProcessEvent(result.ETag, processFunction, _idempotencyEventRepository, cancellationToken);
                     return commandResult.Match(
                         Ok,
                         Problem

src/Altinn.Broker.API/Helpers/WebhookEventHelper.cs → src/Altinn.Broker.API/Helpers/IdempotencyEventHelper.cs

@@ -1,30 +1,31 @@
 using Altinn.Broker.Application;
+using Altinn.Broker.Application.ConfirmDownloadCommand;
 using Altinn.Broker.Core.Domain;
 using Altinn.Broker.Core.Repositories;
 
 using OneOf;
 
 namespace Altinn.Broker.Helpers;
 
-public class WebhookEventHelper
+public class IdempotencyEventHelper
 {
-    public static async Task<OneOf<Task, Error>> ProcessMalwareEvent(ScanResultData data, MalwareScanningResultHandler handler, IWebhookEventRepository webhookEventRepository, CancellationToken cancellationToken)
+    public static async Task<OneOf<Task, Error>> ProcessEvent(string uniqueString, Func<Task<OneOf<Task, Error>>> process, IIdempotencyEventRepository idempotencyEventRepository, CancellationToken cancellationToken)
     {
         try
         {
 
             // Create a new entry for that webhook id
-            await webhookEventRepository.AddWebhookEventAsync(data.ETag, cancellationToken);
+            await idempotencyEventRepository.AddIdempotencyEventAsync(uniqueString, cancellationToken);
             try
             {
                 // Call you method
-                return await handler.Process(data, cancellationToken);
+                return await process();
             }
             catch (Exception e)
             {
                 // Delete the entry on error to make sure the next one isn't ignored
-                await webhookEventRepository.DeleteWebhookEventAsync(data.ETag, cancellationToken);
-                return Task.CompletedTask;
+                await idempotencyEventRepository.DeleteIdempotencyEventAsync(uniqueString, cancellationToken);
+                throw;
             }
         }
         catch (Npgsql.PostgresException e)

src/Altinn.Broker.API/Program.cs

@@ -78,7 +78,7 @@ static void BuildAndRun(string[] args)
     app.MapControllers();
 
     app.UseHangfireDashboard();
-    app.Services.GetService<IRecurringJobManager>().AddOrUpdate<MalwareScanningResultHandler>("Delete old webhook events", handler => handler.DeleteOldWebhookEvents(), Cron.Weekly());
+    app.Services.GetService<IRecurringJobManager>().AddOrUpdate<IdempotencyService>("Delete old impotency events", handler => handler.DeleteOldIdempotencyEvents(), Cron.Weekly());
 
     app.Run();
 }

src/Altinn.Broker.Application/ConfirmDownloadCommand/ConfirmDownloadCommandHandler.cs

@@ -57,6 +57,10 @@ public async Task<OneOf<Task, Error>> Process(ConfirmDownloadCommandRequest requ
         {
             return Errors.FileNotPublished;
         }
+        if (file.RecipientCurrentStatuses.First(recipientStatus => recipientStatus.Actor.ActorExternalId == request.Token.Consumer).Status == ActorFileStatus.DownloadConfirmed)
+        {
+            return Task.CompletedTask;
+        }
 
         await _actorFileStatusRepository.InsertActorFileStatus(request.FileId, ActorFileStatus.DownloadConfirmed, request.Token.Consumer, cancellationToken);
         await _eventBus.Publish(AltinnEventType.DownloadConfirmed, file.ResourceId, file.FileId.ToString(), cancellationToken);

src/Altinn.Broker.Application/Errors.cs

@@ -18,4 +18,5 @@ public static class Errors
     public static Error UploadFailed = new Error(10, "Error occurred while uploading file. See /details for more information.", HttpStatusCode.InternalServerError);
     public static Error ChecksumMismatch = new Error(11, "The checksum of uploaded file did not match the checksum specified in initialize call.", HttpStatusCode.BadRequest);
     public static Error FileNotPublished = new Error(12, "A file can only be confirmed to be downloaded when it is published. See file status.", HttpStatusCode.BadRequest);
+    public static Error FileAlreadyConfirmed = new Error(13, "The file has already been confirmed to be downloaded.", HttpStatusCode.BadRequest);
 }

src/Altinn.Broker.Application/MalwareScanResults/MalwareScanResultHandler.cs

@@ -15,7 +15,7 @@ namespace Altinn.Broker.Application;
 public class MalwareScanningResultHandler : IHandler<ScanResultData, Task>
 {
     private readonly IFileStatusRepository _fileStatusRepository;
-    private readonly IWebhookEventRepository _webhookEventRepository;
+    private readonly IIdempotencyEventRepository _idempotencyEventRepository;
     private readonly IFileRepository _fileRepository;
     private readonly IBackgroundJobClient _backgroundJobClient;
     private readonly IEventBus _eventBus;
@@ -26,14 +26,14 @@ public MalwareScanningResultHandler(
         IFileRepository fileRepository,
         IBackgroundJobClient backgroundJobClient,
         IEventBus eventBus,
-        IWebhookEventRepository webhookEventRepository,
+        IIdempotencyEventRepository idempotencyEventRepository,
         ILogger<MalwareScanningResultHandler> logger)
     {
         _fileStatusRepository = fileStatusRepository;
         _fileRepository = fileRepository;
         _backgroundJobClient = backgroundJobClient;
         _eventBus = eventBus;
-        _webhookEventRepository = webhookEventRepository;
+        _idempotencyEventRepository = idempotencyEventRepository;
         _logger = logger;
     }
 
@@ -64,11 +64,4 @@ public async Task<OneOf<Task, Error>> Process(ScanResultData data, CancellationT
             return Task.CompletedTask;
         }
     }
-    //Tells hangfire to avoid retries on failure
-    [AutomaticRetry(Attempts = 0)]
-    public async Task DeleteOldWebhookEvents()
-    {
-        _logger.LogInformation("Deleting old webhook events");
-        await _webhookEventRepository.DeleteOldWebhookEvents();
-    }
 }

src/Altinn.Broker.Core/Repositories/IIdempotencyEventRepository.cs

@@ -0,0 +1,9 @@
+namespace Altinn.Broker.Core.Repositories
+{
+    public interface IIdempotencyEventRepository
+    {
+        Task AddIdempotencyEventAsync(string id, CancellationToken cancellationToken);
+        Task DeleteIdempotencyEventAsync(string id, CancellationToken cancellationToken);
+        Task DeleteOldIdempotencyEvents();
+    }
+}

src/Altinn.Broker.Integrations/DependencyInjection.cs

@@ -19,7 +19,7 @@ public static void AddIntegrations(this IServiceCollection services)
         services.AddSingleton<IFileStore, BlobService>();
         services.AddScoped<IResourceRepository, AltinnResourceRegistryRepository>();
         services.AddScoped<IAuthorizationService, AltinnAuthorizationService>();
-        services.AddScoped<IWebhookEventRepository, WebhookEventRepository>();
+        services.AddScoped<IIdempotencyEventRepository, IdempotencyEventRepository>();
         services.AddScoped<IEventBus, AltinnEventBus>();
     }
 }

src/Altinn.Broker.Integrations/Hangfire/IdempotencyService.cs

@@ -0,0 +1,28 @@
+using System.Threading.Tasks;
+
+using Altinn.Broker.Core.Repositories;
+
+using Hangfire;
+
+using Microsoft.Extensions.Logging;
+
+public class IdempotencyService
+{
+    private readonly IIdempotencyEventRepository _idempotencyEventRepository;
+    private readonly ILogger<IdempotencyService> _logger;
+
+    public IdempotencyService(
+        IIdempotencyEventRepository idempotencyEventRepository,
+        ILogger<IdempotencyService> logger)
+    {
+        _idempotencyEventRepository = idempotencyEventRepository;
+        _logger = logger;
+    }
+
+    [AutomaticRetry(Attempts = 0)]
+    public async Task DeleteOldIdempotencyEvents()
+    {
+        _logger.LogInformation("Deleting old idempotency events");
+        await _idempotencyEventRepository.DeleteOldIdempotencyEvents();
+    }
+}

src/Altinn.Broker.Persistence/DependencyInjection.cs

@@ -14,6 +14,6 @@ public static void AddPersistence(this IServiceCollection services)
         services.AddSingleton<IFileStatusRepository, FileStatusRepository>();
         services.AddSingleton<IActorFileStatusRepository, ActorFileStatusRepository>();
         services.AddSingleton<IResourceOwnerRepository, ResourceOwnerRepository>();
-        services.AddSingleton<IWebhookEventRepository, WebhookEventRepository>();
+        services.AddSingleton<IIdempotencyEventRepository, IdempotencyEventRepository>();
     }
 }

src/Altinn.Broker.Persistence/Migrations/V0003__idempotency_event.sql

@@ -0,0 +1,4 @@
+CREATE TABLE broker.idempotency_event (
+    idempotency_event_id_pk character varying(80) PRIMARY KEY,
+    created timestamp without time zone NOT NULL
+);

src/Altinn.Broker.Persistence/Migrations/V0004__add_resourcename.sql

@@ -1,2 +1,2 @@
 ALTER TABLE broker.resource_owner
-ADD resource_group_name uuid NOT NULL;
+ADD resource_group_name uuid NOT NULL;

src/Altinn.Broker.Persistence/Repositories/IdempotencyEventRepository.cs

@@ -0,0 +1,46 @@
+using Altinn.Broker.Core.Repositories;
+
+using Npgsql;
+
+namespace Altinn.Broker.Persistence.Repositories;
+
+public class IdempotencyEventRepository : IIdempotencyEventRepository
+{
+    private DatabaseConnectionProvider _connectionProvider;
+
+    public IdempotencyEventRepository(DatabaseConnectionProvider connectionProvider)
+    {
+        _connectionProvider = connectionProvider;
+    }
+
+
+    public async Task AddIdempotencyEventAsync(string IdempotencyEventId, CancellationToken cancellationToken)
+    {
+        NpgsqlCommand command = await _connectionProvider.CreateCommand(
+                    "INSERT INTO broker.idempotency_event (idempotency_event_id_pk, created)" +
+                    "VALUES (@idempotency_event_id_pk, @created) ");
+        command.Parameters.AddWithValue("@idempotency_event_id_pk", IdempotencyEventId);
+        command.Parameters.AddWithValue("@created", DateTime.UtcNow);
+
+        await command.ExecuteNonQueryAsync(cancellationToken);
+    }
+    public async Task DeleteIdempotencyEventAsync(string IdempotencyEventId, CancellationToken cancellationToken)
+    {
+        NpgsqlCommand command = await _connectionProvider.CreateCommand(
+                    "DELETE FROM broker.idempotency_event " +
+                    "WHERE idempotency_event_id_pk = @idempotency_event_id_pk");
+        command.Parameters.AddWithValue("@idempotency_event_id_pk", IdempotencyEventId);
+
+        await command.ExecuteNonQueryAsync(cancellationToken);
+    }
+    public async Task DeleteOldIdempotencyEvents()
+    {
+        NpgsqlCommand command = await _connectionProvider.CreateCommand(
+                    "DELETE FROM broker.idempotency_event " +
+                    "WHERE created < @created");
+
+        command.Parameters.AddWithValue("@created", DateTime.UtcNow.AddDays(-1));
+
+        await command.ExecuteNonQueryAsync();
+    }
+}

start.ps1

@@ -0,0 +1,2 @@
+docker compose up -d
+dotnet watch --project ./src/Altinn.Broker.API/Altinn.Broker.API.csproj