Migrate existing files to profile versions

FileProcessor.BusinessLogic.Tests/FileProfileVersionMigrationServiceTests.cs

@@ -0,0 +1,176 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using FileProcessor.BusinessLogic.Managers;
+using FileProcessor.BusinessLogic.Services;
+using FileProcessor.FileAggregate;
+using FileProcessor.Models;
+using Imposter.Abstractions;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Shared.DomainDrivenDesign.EventSourcing;
+using Shared.EntityFramework;
+using Shared.EventStore.Aggregate;
+using Shouldly;
+using SimpleResults;
+using TransactionProcessor.Database.Contexts;
+using Xunit;
+using FileProfileModel = global::FileProcessor.Models.FileProfile;
+using FileAggregateRoot = FileProcessor.FileAggregate.FileAggregate;
+
+namespace FileProcessor.BusinessLogic.Tests;
+
+public class FileProfileVersionMigrationServiceTests
+{
+    [Fact]
+    public async Task RunAsync_AssignsInitialVersionToLegacyFile()
+    {
+        (EstateManagementContext context, IDbContextResolverImposter<EstateManagementContext> resolver) = this.CreateContext();
+        TransactionProcessor.Database.Entities.File file = FileProcessor.Testing.TestData.Files1[0];
+        context.Files.Add(file);
+        await context.SaveChangesAsync();
+
+        IFileProfileManagerImposter profileManager = new();
+        profileManager.GetAllFileProfileVersions(Arg<CancellationToken>.Any())
+                      .ReturnsAsync(Result.Success(new List<FileProfileModel>
+                      {
+                          new FileProfileModel(file.FileProfileId,
+                                               "Profile",
+                                               "C:/profile",
+                                               "Request",
+                                               "Operator",
+                                               "\n",
+                                               "Handler",
+                                               FileProcessor.Testing.TestData.FileProfileVersionId,
+                                               1)
+                      }));
+
+        FileAggregateRoot aggregate = FileAggregateRoot.Create(file.FileId);
+        aggregate.CreateFile(new FileAggregateExtensions.CreateFileRequest(file.FileImportLogId,
+                                                                            FileProcessor.Testing.TestData.EstateId,
+                                                                            file.MerchantId,
+                                                                            file.UserId,
+                                                                            file.FileProfileId,
+                                                                            file.FileLocation,
+                                                                            FileProcessor.Testing.TestData.FileUploadedDateTime,
+                                                                            FileProcessor.Testing.TestData.SafaricomOperatorId));
+        IAggregateRepositoryImposter<FileAggregateRoot, DomainEvent> repository = new();
+        repository.GetLatestVersion(file.FileId, Arg<CancellationToken>.Any()).ReturnsAsync(Result.Success(aggregate));
+        repository.SaveChanges(Arg<FileAggregateRoot>.Any(), Arg<CancellationToken>.Any()).ReturnsAsync(Result.Success());
+
+        FileProfileVersionMigrationService service = new(profileManager.Instance(),
+                                                          resolver.Instance(),
+                                                          repository.Instance());
+
+        Result<FileProfileVersionMigrationReport> result = await service.RunAsync(CancellationToken.None);
+
+        result.IsSuccess.ShouldBeTrue();
+        result.Data.FilesAssigned.ShouldBe(1);
+        result.Data.Failures.ShouldBeEmpty();
+        aggregate.GetFile().FileProfileVersionId.ShouldBe(FileProcessor.Testing.TestData.FileProfileVersionId);
+    }
+
+    [Fact]
+    public async Task RunAsync_WhenProfileCannotBeResolved_ReportsFailureAndLeavesFileUnchanged()
+    {
+        (EstateManagementContext context, IDbContextResolverImposter<EstateManagementContext> resolver) = this.CreateContext();
+        TransactionProcessor.Database.Entities.File file = FileProcessor.Testing.TestData.Files1[0];
+        file.FileProfileId = Guid.NewGuid();
+        context.Files.Add(file);
+        await context.SaveChangesAsync();
+
+        IFileProfileManagerImposter profileManager = new();
+        profileManager.GetAllFileProfileVersions(Arg<CancellationToken>.Any())
+                      .ReturnsAsync(Result.Success(new List<FileProfileModel>()));
+
+        FileAggregateRoot aggregate = FileAggregateRoot.Create(file.FileId);
+        aggregate.CreateFile(new FileAggregateExtensions.CreateFileRequest(file.FileImportLogId,
+                                                                            FileProcessor.Testing.TestData.EstateId,
+                                                                            file.MerchantId,
+                                                                            file.UserId,
+                                                                            file.FileProfileId,
+                                                                            file.FileLocation,
+                                                                            FileProcessor.Testing.TestData.FileUploadedDateTime,
+                                                                            FileProcessor.Testing.TestData.SafaricomOperatorId));
+        IAggregateRepositoryImposter<FileAggregateRoot, DomainEvent> repository = new();
+        repository.GetLatestVersion(file.FileId, Arg<CancellationToken>.Any()).ReturnsAsync(Result.Success(aggregate));
+
+        FileProfileVersionMigrationService service = new(profileManager.Instance(),
+                                                          resolver.Instance(),
+                                                          repository.Instance());
+
+        Result<FileProfileVersionMigrationReport> result = await service.RunAsync(CancellationToken.None);
+
+        result.IsSuccess.ShouldBeTrue();
+        result.Data.FilesAssigned.ShouldBe(0);
+        result.Data.Failures.Count.ShouldBe(1);
+        result.Data.Failures.Single().ShouldContain(file.FileId.ToString());
+        aggregate.GetFile().FileProfileVersionId.ShouldBe(Guid.Empty);
+    }
+
+    [Fact]
+    public async Task RunAsync_WhenRepeated_SkipsAlreadyAssignedFile()
+    {
+        (EstateManagementContext context, IDbContextResolverImposter<EstateManagementContext> resolver) = this.CreateContext();
+        TransactionProcessor.Database.Entities.File file = FileProcessor.Testing.TestData.Files1[0];
+        context.Files.Add(file);
+        await context.SaveChangesAsync();
+
+        IFileProfileManagerImposter profileManager = new();
+        profileManager.GetAllFileProfileVersions(Arg<CancellationToken>.Any())
+                      .ReturnsAsync(Result.Success(new List<FileProfileModel>
+                      {
+                          new FileProfileModel(file.FileProfileId,
+                                               "Profile",
+                                               "C:/profile",
+                                               "Request",
+                                               "Operator",
+                                               "\n",
+                                               "Handler",
+                                               FileProcessor.Testing.TestData.FileProfileVersionId,
+                                               1)
+                      }));
+
+        FileAggregateRoot aggregate = FileAggregateRoot.Create(file.FileId);
+        aggregate.CreateFile(new FileAggregateExtensions.CreateFileRequest(file.FileImportLogId,
+                                                                            FileProcessor.Testing.TestData.EstateId,
+                                                                            file.MerchantId,
+                                                                            file.UserId,
+                                                                            file.FileProfileId,
+                                                                            file.FileLocation,
+                                                                            FileProcessor.Testing.TestData.FileUploadedDateTime,
+                                                                            FileProcessor.Testing.TestData.SafaricomOperatorId));
+        IAggregateRepositoryImposter<FileAggregateRoot, DomainEvent> repository = new();
+        repository.GetLatestVersion(file.FileId, Arg<CancellationToken>.Any()).ReturnsAsync(Result.Success(aggregate));
+        repository.SaveChanges(Arg<FileAggregateRoot>.Any(), Arg<CancellationToken>.Any()).ReturnsAsync(Result.Success());
+
+        FileProfileVersionMigrationService service = new(profileManager.Instance(),
+                                                          resolver.Instance(),
+                                                          repository.Instance());
+
+        Result<FileProfileVersionMigrationReport> firstResult = await service.RunAsync(CancellationToken.None);
+        Result<FileProfileVersionMigrationReport> secondResult = await service.RunAsync(CancellationToken.None);
+
+        firstResult.Data.FilesAssigned.ShouldBe(1);
+        secondResult.Data.FilesAssigned.ShouldBe(0);
+        secondResult.Data.FilesSkipped.ShouldBe(1);
+        secondResult.Data.Failures.ShouldBeEmpty();
+    }
+
+    private (EstateManagementContext Context, IDbContextResolverImposter<EstateManagementContext> Resolver) CreateContext()
+    {
+        DbContextOptions<EstateManagementContext> options = new DbContextOptionsBuilder<EstateManagementContext>()
+            .UseInMemoryDatabase(Guid.NewGuid().ToString("N"))
+            .Options;
+        EstateManagementContext context = new(options);
+        ServiceCollection services = new();
+        services.AddTransient(_ => context);
+        ServiceProvider provider = services.BuildServiceProvider();
+        IServiceScope scope = provider.CreateScope();
+        IDbContextResolverImposter<EstateManagementContext> resolver = new();
+        resolver.Resolve(Arg<string>.Any(), Arg<string>.Any()).Returns(new ResolvedDbContext<EstateManagementContext>(scope));
+        return (context, resolver);
+    }
+}

FileProcessor.BusinessLogic/Services/FileProfileVersionMigrationService.cs

@@ -0,0 +1,141 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using FileProcessor.BusinessLogic.Managers;
+using FileProcessor.FileAggregate;
+using FileProcessor.Models;
+using Microsoft.EntityFrameworkCore;
+using Shared.DomainDrivenDesign.EventSourcing;
+using Shared.EntityFramework;
+using Shared.EventStore.Aggregate;
+using Shared.Results;
+using SimpleResults;
+using TransactionProcessor.Database.Contexts;
+using FileAggregateRoot = FileProcessor.FileAggregate.FileAggregate;
+using FileProfileModel = global::FileProcessor.Models.FileProfile;
+
+namespace FileProcessor.BusinessLogic.Services;
+
+public interface IFileProfileVersionMigrationService
+{
+    Task<Result<FileProfileVersionMigrationReport>> RunAsync(CancellationToken cancellationToken);
+}
+
+public sealed class FileProfileVersionMigrationService : IFileProfileVersionMigrationService
+{
+    private const String EstateManagementDatabaseName = "TransactionProcessorReadModel";
+
+    private readonly IFileProfileManager FileProfileManager;
+    private readonly IDbContextResolver<EstateManagementContext> Resolver;
+    private readonly IAggregateRepository<FileAggregateRoot, DomainEvent> FileAggregateRepository;
+
+    public FileProfileVersionMigrationService(IFileProfileManager fileProfileManager,
+                                               IDbContextResolver<EstateManagementContext> resolver,
+                                               IAggregateRepository<FileAggregateRoot, DomainEvent> fileAggregateRepository)
+    {
+        this.FileProfileManager = fileProfileManager;
+        this.Resolver = resolver;
+        this.FileAggregateRepository = fileAggregateRepository;
+    }
+
+    public async Task<Result<FileProfileVersionMigrationReport>> RunAsync(CancellationToken cancellationToken)
+    {
+        Result<List<FileProfileModel>> profileVersionsResult = await this.FileProfileManager.GetAllFileProfileVersions(cancellationToken);
+        if (profileVersionsResult.IsFailed)
+            return ResultHelpers.CreateFailure(profileVersionsResult);
+
+        FileProfileVersionMigrationReport report = new();
+        Dictionary<Guid, Guid> initialVersions = this.GetInitialVersions(profileVersionsResult.Data, report);
+
+        ResolvedDbContext<EstateManagementContext> resolvedContext = this.Resolver.Resolve(EstateManagementDatabaseName, null);
+        List<ExistingFile> files = await resolvedContext.Context.Files
+                                                        .AsNoTracking()
+                                                        .Select(file => new ExistingFile(file.FileId, file.FileProfileId))
+                                                        .ToListAsync(cancellationToken);
+
+        foreach (ExistingFile file in files)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            report.FilesChecked++;
+
+            Result<FileAggregateRoot> aggregateResult = await this.FileAggregateRepository.GetLatestVersion(file.FileId, cancellationToken);
+            if (aggregateResult.IsFailed || aggregateResult.Data == null)
+            {
+                report.Failures.Add($"File [{file.FileId}] could not be loaded: {aggregateResult.Message}");
+                continue;
+            }
+
+            FileDetails fileDetails = aggregateResult.Data.GetFile();
+            if (fileDetails.FileProfileVersionId != Guid.Empty)
+            {
+                report.FilesSkipped++;
+                continue;
+            }
+
+            if (initialVersions.TryGetValue(file.FileProfileId, out Guid initialVersionId) == false)
+            {
+                report.Failures.Add($"File [{file.FileId}] references profile [{file.FileProfileId}], which has no usable initial version");
+                continue;
+            }
+
+            Result assignResult = aggregateResult.Data.AssignFileProfileVersion(initialVersionId);
+            if (assignResult.IsFailed)
+            {
+                report.Failures.Add($"File [{file.FileId}] could not be assigned profile version [{initialVersionId}]: {assignResult.Message}");
+                continue;
+            }
+
+            Result saveResult = await this.FileAggregateRepository.SaveChanges(aggregateResult.Data, cancellationToken);
+            if (saveResult.IsFailed)
+            {
+                report.Failures.Add($"File [{file.FileId}] could not be saved: {saveResult.Message}");
+                continue;
+            }
+
+            report.FilesAssigned++;
+        }
+
+        return Result.Success(report);
+    }
+
+    private Dictionary<Guid, Guid> GetInitialVersions(List<FileProfileModel> profileVersions,
+                                                       FileProfileVersionMigrationReport report)
+    {
+        Dictionary<Guid, Guid> initialVersions = new();
+
+        foreach (IGrouping<Guid, FileProfileModel> profileGroup in profileVersions.GroupBy(profile => profile.FileProfileId))
+        {
+            report.ProfilesChecked++;
+            List<FileProfileModel> versions = profileGroup.Where(profile => profile.FileProfileVersion == 1 && profile.FileProfileVersionId != Guid.Empty).ToList();
+            if (versions.Count != 1)
+            {
+                report.Failures.Add($"Profile [{profileGroup.Key}] has {versions.Count} usable initial versions");
+                continue;
+            }
+
+            initialVersions[profileGroup.Key] = versions[0].FileProfileVersionId;
+            report.ProfilesWithInitialVersion++;
+        }
+
+        return initialVersions;
+    }
+
+    private sealed record ExistingFile(Guid FileId, Guid FileProfileId);
+}
+
+public sealed class FileProfileVersionMigrationReport
+{
+    public Int32 ProfilesChecked { get; set; }
+
+    public Int32 ProfilesWithInitialVersion { get; set; }
+
+    public Int32 FilesChecked { get; set; }
+
+    public Int32 FilesAssigned { get; set; }
+
+    public Int32 FilesSkipped { get; set; }
+
+    public List<String> Failures { get; } = new();
+}

FileProcessor.File.DomainEvents/FileAggregateDomainEvents.cs.cs

@@ -30,6 +30,9 @@ public record FileCreatedEvent(Guid FileId,
                                    Guid OperatorId,
                                    Guid FileProfileVersionId = default) : DomainEvent(FileId, Guid.NewGuid());
 
+    public record FileProfileVersionAssignedEvent(Guid FileId,
+                                                   Guid FileProfileVersionId) : DomainEvent(FileId, Guid.NewGuid());
+
     public record FileRejectedEvent(Guid FileId,
                                     Guid EstateId,
                                     Guid MerchantId,

FileProcessor.FileAggregate.Tests/FileAggregateTests.cs

@@ -13,6 +13,44 @@ namespace FileProcessor.FileAggregate.Tests
 
     public class FileAggregateTests
     {
+        [Fact]
+        public void FileAggregate_AssignProfileVersion_WhenFileHasNoVersion_AssignsVersion()
+        {
+            FileAggregate aggregate = FileAggregate.Create(TestData.FileId);
+            aggregate.CreateFile(new FileAggregateExtensions.CreateFileRequest(TestData.FileImportLogId,
+                                                                                TestData.EstateId,
+                                                                                TestData.MerchantId,
+                                                                                TestData.UserId,
+                                                                                TestData.FileProfileId,
+                                                                                TestData.FileLocation,
+                                                                                TestData.FileUploadedDateTime,
+                                                                                TestData.SafaricomOperatorId));
+
+            Result result = aggregate.AssignFileProfileVersion(TestData.FileProfileVersionId);
+
+            result.IsSuccess.ShouldBeTrue();
+            aggregate.GetFile().FileProfileVersionId.ShouldBe(TestData.FileProfileVersionId);
+        }
+
+        [Fact]
+        public void FileAggregate_ReplayAssignedProfileVersionEvent_RestoresVersion()
+        {
+            FileAggregate aggregate = FileAggregate.Create(TestData.FileId);
+            aggregate.PlayEvent(new FileCreatedEvent(TestData.FileId,
+                                                      TestData.FileImportLogId,
+                                                      TestData.EstateId,
+                                                      TestData.MerchantId,
+                                                      TestData.UserId,
+                                                      TestData.FileProfileId,
+                                                      TestData.FileLocation,
+                                                      TestData.FileUploadedDateTime,
+                                                      TestData.SafaricomOperatorId));
+
+            aggregate.PlayEvent(new FileProfileVersionAssignedEvent(TestData.FileId, TestData.FileProfileVersionId));
+
+            aggregate.GetFile().FileProfileVersionId.ShouldBe(TestData.FileProfileVersionId);
+        }
+
         [Fact]
         public void FileAggregate_CanBeCreated_IsCreated()
         {