diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..3382a44 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,16 @@ +version: 2 +updates: + - package-ecosystem: "nuget" + directory: "/" + schedule: + interval: "weekly" + day: "sunday" + time: "04:00" + target-branch: "develop" + commit-message: + prefix: "deps" + open-pull-requests-limit: 1 + groups: + all-dependencies: + patterns: + - "*" diff --git a/.github/workflows/build-artifacts.yml b/.github/workflows/build-artifacts.yml index c429830..bbb716d 100644 --- a/.github/workflows/build-artifacts.yml +++ b/.github/workflows/build-artifacts.yml @@ -23,7 +23,7 @@ jobs: run: dotnet restore - name: Publish - run: dotnet publish -c Release --artifacts-path ./output + run: dotnet publish -c Release -r win-x64 --artifacts-path ./output - name: Upload build artifacts uses: actions/upload-artifact@v4 diff --git a/.gitignore b/.gitignore index aa3f6cb..373dbd4 100644 --- a/.gitignore +++ b/.gitignore @@ -280,6 +280,9 @@ LocalFiles/ *.dat *~lock.* *.pubxml +cfoextract* +C_NOMIS_OFFENDER* +**/DMS_STAGING/ # Mac .DS_STORE \ No newline at end of file diff --git a/Directory.Packages.props b/Directory.Packages.props index ead8596..510ccdc 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -3,63 +3,51 @@ true - - - - - - - + + + + + + + + + - - - + + all runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - - - + + + + - - + + all runtime; build; native; contentfiles; analyzers; buildtransitive - - - - - - - - - - - - + + + + + + + + + + - - - - - - - + + @@ -67,22 +55,17 @@ - - - + - + - - - all diff --git a/README.md b/README.md index 234a2c3..379db16 100644 --- a/README.md +++ b/README.md @@ -12,11 +12,7 @@ Any queries, please contact andrew.grocott@justice.gov.uk or visit our slack cha # Development Setup and Execution Guide ## Setup (development) -1. From the **project root**, run the setup script. This will automatically create the required folder structure in your devices home directory: - ```sh - ./setup.sh - ``` -2. Configure secret(s) for applications in the *src* directory: +1. To use the Visualiser app, you must configure secret(s) for applications in the *src* directory: - *Visualiser.csproj* → Manage User Secrets ```json { @@ -28,7 +24,3 @@ Any queries, please contact andrew.grocott@justice.gov.uk or visit our slack cha The recommended way to run and debug these apps is using .NET Aspire. - **Using Visual Studio Code**: open the project and press `F5`, selecting the *Default Configuration*. - **Using Visual Studio or other IDEs**: From the debug configuration dropdown, select `Aspire.AppHost` and start the application. - -### Known issues -1. Due to a [known issue](https://github.com/CommunityToolkit/Aspire/issues/942) in the Aspire Community Toolkit, some SQL projects may start too early. - - You may need to re-run the project for database changes to apply correctly. diff --git a/base.development.env b/base.development.env index 56aec7c..8d6d6ea 100644 --- a/base.development.env +++ b/base.development.env @@ -1,11 +1,3 @@ DOTNET_ENVIRONMENT=Development -OFFLOC_USERNAME= -OFFLOC_PASSWORD= -DELIUS_USERNAME= -DELIUS_PASSWORD= -Staging__DownloadOfflocFiles=false -Staging__DownloadDeliusFiles=false - -Parallel=false RedundantOfflocFields=11,20,32,33,34,35,36,37,38,41,43,44,45,46,53,54,55,56,57,58,59,60,61,62,74,75,76,117,118,119,120,121,122,123,124,125,126,127,128,129,130,131,132,142,144 \ No newline at end of file diff --git a/development.docker.env b/development.docker.env deleted file mode 100644 index 86138fa..0000000 --- a/development.docker.env +++ /dev/null @@ -1,37 +0,0 @@ -# extends: .env.development.base - -DOTNET_ENVIRONMENT=Development - -# Core -OFFLOC_USERNAME= -OFFLOC_PASSWORD= -DELIUS_USERNAME= -DELIUS_PASSWORD= -NEXTCLOUD_ADDRESS= -Staging__DownloadOfflocFiles=false -Staging__DownloadDeliusFiles=false - -Parallel=false -RedundantOfflocFields=11,20,32,33,34,35,36,37,38,41,43,44,45,46,53,54,55,56,57,58,59,60,61,62,74,75,76,117,118,119,120,121,122,123,124,125,126,127,128,129,130,131,132,142,144 - -# Rabbit -RABBIT_HOSTING=host.docker.internal -RABBIT_USERNAME=dms -RABBIT_PASSWORD=R@bbit - -# Connection Strings -USER_ID=sa -SQL_HOST=host.docker.internal,1433 -SA_PASSWORD=Pass@word -ConnectionStrings__ClusterDb=Data Source=${SQL_HOST};Initial Catalog=ClusterDb;User Id=${USER_ID};Password=${SA_PASSWORD};TrustServerCertificate=true -ConnectionStrings__DeliusRunningPictureDb=Data Source=${SQL_HOST};Initial Catalog=DeliusRunningPictureDb;User Id=${USER_ID};Password=${SA_PASSWORD};TrustServerCertificate=true -ConnectionStrings__DeliusStagingDb=Data Source=${SQL_HOST};Initial Catalog=DeliusStagingDb;User Id=${USER_ID};Password=${SA_PASSWORD};TrustServerCertificate=true -ConnectionStrings__MatchingDb=Data Source=${SQL_HOST};Initial Catalog=MatchingDb;User Id=${USER_ID};Password=${SA_PASSWORD};TrustServerCertificate=true -ConnectionStrings__OfflocRunningPictureDb=Data Source=${SQL_HOST};Initial Catalog=OfflocRunningPictureDb;User Id=${USER_ID};Password=${SA_PASSWORD};TrustServerCertificate=true -ConnectionStrings__OfflocStagingDb=Data Source=${SQL_HOST};Initial Catalog=OfflocStagingDb;User Id=${USER_ID};Password=${SA_PASSWORD};TrustServerCertificate=true - -#DMSFilesBasePath= #Not used here. -DeliusInputMount=deliusInput -DeliusOutputMount=deliusOutput -OfflocInputMount=offlocInput -OfflocOutputMount=offlocOutput \ No newline at end of file diff --git a/development.local.env b/development.local.env index 70e906c..c7ac801 100644 --- a/development.local.env +++ b/development.local.env @@ -2,16 +2,6 @@ DOTNET_ENVIRONMENT=Development -# Core -OFFLOC_USERNAME= -OFFLOC_PASSWORD= -DELIUS_USERNAME= -DELIUS_PASSWORD= -NEXTCLOUD_ADDRESS= -Staging__DownloadOfflocFiles=false -Staging__DownloadDeliusFiles=false - -Parallel=false RedundantOfflocFields= Age, Occupation Description, Check Hold Governor, Check Hold General (to be left blank), Check Hold Discipline (to be left blank), Check Hold Allocation, Check Hold Security, Check Hold Medical, Check Hold Parole, ACCT Status (F2052), Status Rank (to be left blank), Pending Transfers (Full Establishment Name), Received From, Vulnerable Prisoner Alert, Height (metres), Complexion, Hair Colour, Left Eye, Right Eye, Build, Facial Shape, Facial Hair, Physical Mark Head, Physical Mark Body, Rule 45/YOI Rule 49, ACCT (Self Harm) Status, ACCT (Self Harm) Start Date, Remark Type Allocation, Remarks Allocation, Remark Type Security, Remarks Security, Remark Type Medical, Remarks Medical, Remark Type Parole, Remarks Parole, Remark Type Discipline, Remarks Discipline, Remark Type General, Remarks General, Remark Type Reception, Remarks Reception, Remark Type Labour, Remarks Labour, Date Of First Movement, Diary Details # Rabbit @@ -62,8 +52,5 @@ Serilog__Enrich__1=WithMachineName Serilog__Enrich__2=WithProcessId Serilog__Enrich__3=WithThreadId -DMSFilesBasePath=~/DMS/ -DeliusInputMount=${DMSFilesBasePath}\Delius\Input -DeliusOutputMount=${DMSFilesBasePath}\Delius\Output -OfflocInputMount=${DMSFilesBasePath}\Offloc\Input -OfflocOutputMount=${DMSFilesBasePath}\Offloc\Output \ No newline at end of file +# e.g. /app/ +#DMSFilesBasePath=~/DMS/ \ No newline at end of file diff --git a/dms.sln b/dms.sln index 82e9717..caf5feb 100644 --- a/dms.sln +++ b/dms.sln @@ -19,8 +19,6 @@ Project("{42EA0DBD-9CF1-443E-919E-BE9C484E4577}") = "OfflocStagingDb", "src\Offl EndProject Project("{42EA0DBD-9CF1-443E-919E-BE9C484E4577}") = "DeliusStagingDb", "src\DeliusStagingDb\DeliusStagingDb.sqlproj", "{5E86F978-BE47-458C-ADE7-D3B5B47C514A}" EndProject -Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FileStorage", "src\FileStorage\FileStorage.csproj", "{6783146D-6C22-4332-861D-81A1D4DA1CFD}" -EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Cleanup", "src\Cleanup\Cleanup.csproj", "{4C0C21B2-853F-44FD-A867-66AC42F18665}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Delius", "Delius", "{581C4EBC-8973-4D51-BD8E-F76AA4E16773}" @@ -74,8 +72,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "API", "src\API\API.csproj", EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Api.Tests", "src\Api.Tests\Api.Tests.csproj", "{A4247B60-97A2-4D5A-AE97-53E3D497675A}" EndProject -Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Orchestrator", "src\Orchestrator\Orchestrator.csproj", "{D0E5214B-119B-4D1B-947F-8A99F7AE65E2}" -EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Aspire.AppHost", "src\Aspire\Aspire.AppHost\Aspire.AppHost.csproj", "{F00FE6A0-A6B1-4877-B988-69690D11066B}" EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Aspire.ServiceDefaults", "src\Aspire\Aspire.ServiceDefaults\Aspire.ServiceDefaults.csproj", "{257BF0CC-7E4B-44EF-A01D-54BD003EE089}" @@ -90,6 +86,8 @@ Project("{42EA0DBD-9CF1-443E-919E-BE9C484E4577}") = "AuditDb", "src\AuditDb\Audi EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "FakeDataSeeder", "src\FakeDataSeeder\FakeDataSeeder.csproj", "{F7F37423-FFBD-4093-AC69-E1D38B208998}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "FileSync", "src\FileSync\FileSync.csproj", "{74BC61BE-4CAC-4310-8BEA-82135EFE5B07}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -196,18 +194,6 @@ Global {5E86F978-BE47-458C-ADE7-D3B5B47C514A}.Release|x86.ActiveCfg = Release|Any CPU {5E86F978-BE47-458C-ADE7-D3B5B47C514A}.Release|x86.Build.0 = Release|Any CPU {5E86F978-BE47-458C-ADE7-D3B5B47C514A}.Release|x86.Deploy.0 = Release|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Debug|Any CPU.Build.0 = Debug|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Debug|x64.ActiveCfg = Debug|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Debug|x64.Build.0 = Debug|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Debug|x86.ActiveCfg = Debug|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Debug|x86.Build.0 = Debug|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Release|Any CPU.ActiveCfg = Release|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Release|Any CPU.Build.0 = Release|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Release|x64.ActiveCfg = Release|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Release|x64.Build.0 = Release|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Release|x86.ActiveCfg = Release|Any CPU - {6783146D-6C22-4332-861D-81A1D4DA1CFD}.Release|x86.Build.0 = Release|Any CPU {4C0C21B2-853F-44FD-A867-66AC42F18665}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {4C0C21B2-853F-44FD-A867-66AC42F18665}.Debug|Any CPU.Build.0 = Debug|Any CPU {4C0C21B2-853F-44FD-A867-66AC42F18665}.Debug|x64.ActiveCfg = Debug|Any CPU @@ -400,18 +386,6 @@ Global {A4247B60-97A2-4D5A-AE97-53E3D497675A}.Release|x64.Build.0 = Release|Any CPU {A4247B60-97A2-4D5A-AE97-53E3D497675A}.Release|x86.ActiveCfg = Release|Any CPU {A4247B60-97A2-4D5A-AE97-53E3D497675A}.Release|x86.Build.0 = Release|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Debug|Any CPU.Build.0 = Debug|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Debug|x64.ActiveCfg = Debug|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Debug|x64.Build.0 = Debug|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Debug|x86.ActiveCfg = Debug|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Debug|x86.Build.0 = Debug|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Release|Any CPU.ActiveCfg = Release|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Release|Any CPU.Build.0 = Release|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Release|x64.ActiveCfg = Release|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Release|x64.Build.0 = Release|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Release|x86.ActiveCfg = Release|Any CPU - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2}.Release|x86.Build.0 = Release|Any CPU {F00FE6A0-A6B1-4877-B988-69690D11066B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {F00FE6A0-A6B1-4877-B988-69690D11066B}.Debug|Any CPU.Build.0 = Debug|Any CPU {F00FE6A0-A6B1-4877-B988-69690D11066B}.Debug|x64.ActiveCfg = Debug|Any CPU @@ -502,6 +476,18 @@ Global {F7F37423-FFBD-4093-AC69-E1D38B208998}.Release|x64.Build.0 = Release|Any CPU {F7F37423-FFBD-4093-AC69-E1D38B208998}.Release|x86.ActiveCfg = Release|Any CPU {F7F37423-FFBD-4093-AC69-E1D38B208998}.Release|x86.Build.0 = Release|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Debug|Any CPU.Build.0 = Debug|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Debug|x64.ActiveCfg = Debug|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Debug|x64.Build.0 = Debug|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Debug|x86.ActiveCfg = Debug|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Debug|x86.Build.0 = Debug|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Release|Any CPU.ActiveCfg = Release|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Release|Any CPU.Build.0 = Release|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Release|x64.ActiveCfg = Release|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Release|x64.Build.0 = Release|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Release|x86.ActiveCfg = Release|Any CPU + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -514,7 +500,6 @@ Global {47A53A37-1E8C-479C-B3B5-276D2D05154E} = {C4AA533A-1A3C-4BBC-BA2B-A678FB784CA2} {0FA34F76-91ED-4037-883D-5916258BA65A} = {25487F3E-ECA0-4FC3-A61F-13DCC56974D1} {5E86F978-BE47-458C-ADE7-D3B5B47C514A} = {581C4EBC-8973-4D51-BD8E-F76AA4E16773} - {6783146D-6C22-4332-861D-81A1D4DA1CFD} = {605FBD70-9552-462C-BAC2-90D4655F1A27} {4C0C21B2-853F-44FD-A867-66AC42F18665} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} {581C4EBC-8973-4D51-BD8E-F76AA4E16773} = {C4AA533A-1A3C-4BBC-BA2B-A678FB784CA2} {25487F3E-ECA0-4FC3-A61F-13DCC56974D1} = {C4AA533A-1A3C-4BBC-BA2B-A678FB784CA2} @@ -535,7 +520,6 @@ Global {A1A8A7ED-D3D2-4916-866A-9AEB510840AC} = {8242D833-6E75-4EAA-9EB1-57B7EC128354} {EBFD06D5-A7B3-48B7-8646-3155538BC867} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} {A4247B60-97A2-4D5A-AE97-53E3D497675A} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} - {D0E5214B-119B-4D1B-947F-8A99F7AE65E2} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} {F00FE6A0-A6B1-4877-B988-69690D11066B} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} {257BF0CC-7E4B-44EF-A01D-54BD003EE089} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} {F300A5B2-A029-F04C-F6BB-AD41877264E3} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} @@ -543,6 +527,7 @@ Global {C512532D-ADF0-D460-B843-C239A2A093EC} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} {22A80A04-1AC5-46F1-BE57-DE9B1456B34E} = {8242D833-6E75-4EAA-9EB1-57B7EC128354} {F7F37423-FFBD-4093-AC69-E1D38B208998} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} + {74BC61BE-4CAC-4310-8BEA-82135EFE5B07} = {D30CBF2D-71E6-43EE-85EC-BF193B03782D} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {39E62751-A441-4DEF-AD24-20CDC6D59FF3} diff --git a/setup.sh b/setup.sh deleted file mode 100644 index 58d44d8..0000000 --- a/setup.sh +++ /dev/null @@ -1,25 +0,0 @@ -#!/usr/bin/env bash - -# Base directory in home -BASE_DIR="$HOME/DMS" - -# Folder structure array -DIRS=( - "$BASE_DIR/Delius/Input" - "$BASE_DIR/Delius/Output" - "$BASE_DIR/Offloc/Input" - "$BASE_DIR/Offloc/Output" -) - -echo "Creating DMS folder structure..." - -for dir in "${DIRS[@]}"; do - if [ ! -d "$dir" ]; then - mkdir -p "$dir" - echo "📁 Created: $dir" - else - echo "Already exists: $dir" - fi -done - -echo "🎉 Setup complete!" \ No newline at end of file diff --git a/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj b/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj index 9f3d04c..5ed2065 100644 --- a/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj +++ b/src/Aspire/Aspire.AppHost/Aspire.AppHost.csproj @@ -1,12 +1,11 @@  - + Exe enable enable true 4314ae44-7ea1-4bdc-bda3-c55826dd2434 - net9.0 @@ -15,6 +14,7 @@ + @@ -36,6 +36,7 @@ + @@ -43,7 +44,6 @@ - diff --git a/src/Aspire/Aspire.AppHost/Extensions/AppExtensions.cs b/src/Aspire/Aspire.AppHost/Extensions/AppExtensions.cs index 0f4163b..105fc15 100644 --- a/src/Aspire/Aspire.AppHost/Extensions/AppExtensions.cs +++ b/src/Aspire/Aspire.AppHost/Extensions/AppExtensions.cs @@ -29,33 +29,40 @@ public static IResourceBuilder AddDmsVisualiser( public static IDistributedApplicationBuilder AddDmsServices( this IDistributedApplicationBuilder builder, + IResourceBuilder minio, IResourceBuilder rabbit, - DmsDatabaseResources databases) + DmsDatabaseResources databases, + string hostMount, + string targetMount) { - builder.AddDmsService("Blocking", rabbit, databases); - builder.AddDmsService("Cleanup", rabbit, databases); - builder.AddDmsService("DbInteractions", rabbit, databases); - builder.AddDmsService("Delius-Parser", rabbit, databases); - builder.AddDmsService("Import", rabbit, databases); - builder.AddDmsService("Logging", rabbit, databases); - builder.AddDmsService("Matching-Engine", rabbit, databases); - //builder.AddDmsService("Meow", rabbit, databases); - builder.AddDmsService("Offloc-Cleaner", rabbit, databases); - builder.AddDmsService("Offloc-Parser", rabbit, databases); - //builder.AddDmsService("Orchestrator", rabbit, databases); + builder.AddDmsService("Blocking", rabbit, databases, hostMount); + builder.AddDmsService("Cleanup", rabbit, databases, hostMount); - // Entry point - start the app to 'kick off' DMS - builder.AddDmsService("Kickoff", rabbit, databases) - .WithExplicitStart(); + builder.AddDmsService("DbInteractions", rabbit, databases, hostMount) + .WithEnvironment("DmsFilesBasePath", targetMount); // Override default with linux (sql container) path + + builder.AddDmsService("Delius-Parser", rabbit, databases, hostMount); + builder.AddDmsService("Import", rabbit, databases, hostMount); + builder.AddDmsService("Logging", rabbit, databases, hostMount); + builder.AddDmsService("Matching-Engine", rabbit, databases, hostMount); + //builder.AddDmsService("Meow", rabbit, databases, hostMount); + builder.AddDmsService("Offloc-Cleaner", rabbit, databases, hostMount); + builder.AddDmsService("Offloc-Parser", rabbit, databases, hostMount); + builder.AddDmsService("FileSync", rabbit, databases, hostMount) + .WithReference(minio).WaitFor(minio) + .WithExplicitStart() + .WithEnvironment("MinIO:BucketName", "cfo-dms-files"); + return builder; } private static IResourceBuilder AddDmsService( - this IDistributedApplicationBuilder builder, + this IDistributedApplicationBuilder builder, string name, IResourceBuilder rabbit, - DmsDatabaseResources databases) where TDmsProject : IProjectMetadata, new() + DmsDatabaseResources databases, + string hostMount) where TDmsProject : IProjectMetadata, new() { var project = builder.AddProject(name) .WithReference(rabbit).WaitFor(rabbit) @@ -66,7 +73,8 @@ private static IResourceBuilder AddDmsService( .WithDmsDatabaseReference(databases.Audit) .WithDmsDatabaseReference(databases.Cluster) .WithDmsDatabaseReference(databases.Matching) - .WithEnvironment("DmsFilesBasePath", "~/DMS/"); + .WithEnvironment("DmsFilesBasePath", hostMount) + .WithEnvironment("DOTNET_ENVIRONMENT", "Development"); return project; } diff --git a/src/Aspire/Aspire.AppHost/Extensions/DatabaseExtensions.cs b/src/Aspire/Aspire.AppHost/Extensions/DatabaseExtensions.cs index ab27a51..c8d3aac 100644 --- a/src/Aspire/Aspire.AppHost/Extensions/DatabaseExtensions.cs +++ b/src/Aspire/Aspire.AppHost/Extensions/DatabaseExtensions.cs @@ -13,7 +13,6 @@ public static IResourceBuilder AddDmsSqlServer( #pragma warning disable ASPIREPROXYENDPOINTS001 return builder.AddSqlServer("sql", password, 61749) .WithDataVolume("dms-data") - .WithBindMount($"{home}/DMS", "/app/") .WithLifetime(ContainerLifetime.Persistent) .WithEndpointProxySupport(false) .WithImageTag("2022-latest"); diff --git a/src/Aspire/Aspire.AppHost/Extensions/HostExtensions.cs b/src/Aspire/Aspire.AppHost/Extensions/HostExtensions.cs new file mode 100644 index 0000000..8fe64fc --- /dev/null +++ b/src/Aspire/Aspire.AppHost/Extensions/HostExtensions.cs @@ -0,0 +1,15 @@ +namespace Aspire.AppHost.Extensions; + +public static class HostExtensions +{ + public static string Create(string hostDir) + { + var delius = Path.Combine(hostDir, "Delius"); + var offloc = Path.Combine(hostDir, "Offloc"); + Directory.CreateDirectory(Path.Combine(delius, "Input")); + Directory.CreateDirectory(Path.Combine(delius, "Output")); + Directory.CreateDirectory(Path.Combine(offloc, "Input")); + Directory.CreateDirectory(Path.Combine(offloc, "Output")); + return hostDir; + } +} \ No newline at end of file diff --git a/src/Aspire/Aspire.AppHost/Program.cs b/src/Aspire/Aspire.AppHost/Program.cs index b22dfdf..f2bb85f 100644 --- a/src/Aspire/Aspire.AppHost/Program.cs +++ b/src/Aspire/Aspire.AppHost/Program.cs @@ -3,12 +3,19 @@ var builder = DistributedApplication.CreateBuilder(args); // Parameters -var password = builder.AddParameter("password", true); -var apiKey = builder.AddParameter("apikey", true); -var isDevelopment = builder.AddParameter("IsDevelopment"); +var apiKey = builder.AddParameter("apiKey", true); +var isDevelopment = builder.AddParameter("isDevelopment"); +var rabbitPassword = builder.AddParameter("rabbitPassword", true); +var sqlPassword = builder.AddParameter("sqlPassword", true); +var minioPassword = builder.AddParameter("minioPassword", true); + +var hostMount = HostExtensions.Create(Path.Combine(builder.AppHostDirectory, "DMS_STAGING")); +var targetMount = "/app/"; // Database setup -var sql = builder.AddDmsSqlServer(password); +var sql = builder.AddDmsSqlServer(sqlPassword) + .WithBindMount(hostMount, targetMount); + var databases = builder.AddDmsDatabases(sql, seedData: false); // API setup @@ -18,9 +25,18 @@ builder.AddDmsVisualiser(apiService); var rabbit = builder - .AddRabbitMQ("RabbitMQ") + .AddRabbitMQ("RabbitMQ", password: rabbitPassword) .WithManagementPlugin(); -builder.AddDmsServices(rabbit, databases); +// MinIO (s3 emulation) +var minio = builder.AddMinioContainer("minio", rootPassword: minioPassword) + .WithDataVolume("dms-minio-data"); + +builder.AddDmsServices( + minio, + rabbit, + databases, + hostMount, + targetMount); builder.Build().Run(); \ No newline at end of file diff --git a/src/Aspire/Aspire.AppHost/appsettings.json b/src/Aspire/Aspire.AppHost/appsettings.json index e4919bd..168b646 100644 --- a/src/Aspire/Aspire.AppHost/appsettings.json +++ b/src/Aspire/Aspire.AppHost/appsettings.json @@ -7,8 +7,10 @@ } }, "Parameters": { - "password": "P@ssword123!", - "apikey": "password", - "isDevelopment": "True" + "apiKey": "password", + "isDevelopment": "True", + "rabbitPassword": "guest", + "sqlPassword": "P@ssword123!", + "minioPassword": "minioadmin" } } diff --git a/src/Blocking/Blocking.csproj b/src/Blocking/Blocking.csproj index f91144e..f929f58 100644 --- a/src/Blocking/Blocking.csproj +++ b/src/Blocking/Blocking.csproj @@ -21,7 +21,6 @@ - diff --git a/src/Cleanup/Cleanup.csproj b/src/Cleanup/Cleanup.csproj index 17f0f16..9758912 100644 --- a/src/Cleanup/Cleanup.csproj +++ b/src/Cleanup/Cleanup.csproj @@ -17,12 +17,10 @@ - - diff --git a/src/DbInteractions/DbBackgroundService.cs b/src/DbInteractions/DbBackgroundService.cs index a3c8caa..0ee7a08 100644 --- a/src/DbInteractions/DbBackgroundService.cs +++ b/src/DbInteractions/DbBackgroundService.cs @@ -29,68 +29,112 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await Task.CompletedTask; - dbMessagingService.DbLongSubscribe(async(message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { string[] results = await dbInteractionService.GetProcessedDeliusFileNames(); DeliusFilesReturnMessage msg = new DeliusFilesReturnMessage(results); dbMessagingService.DbPublishResponse(msg); }, TDbQueue.GetProcessedDeliusFiles); - dbMessagingService.DbLongSubscribe(async(message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { string[] results = await dbInteractionService.GetProcessedOfflocFileNames(); dbMessagingService.DbPublishResponse(new OfflocFilesReturnMessage(results)); }, TDbQueue.GetProcessedOfflocFiles); - dbMessagingService.DbLongSubscribe(async (message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { await dbInteractionService.StageDelius(message.fileName, message.filePath); dbMessagingService.DbPublishResponse(new StageDeliusReturnMessage()); mergingService.MergingPublish(new DeliusFilesCleanupMessage(message.fileName)); }, TDbQueue.StageDelius); - dbMessagingService.DbLongSubscribe(async (message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { //await dbInteractionService.StandardiseDeliusStaging(); - await dbInteractionService.MergeDeliusPicture(); + await dbInteractionService.MergeDeliusPicture(message.fileName); await dbInteractionService.ClearDeliusStaging(); - dbMessagingService.DbPublishResponse(new MergeDeliusReturnMessage()); - }, TDbQueue.MergeDelius); + dbMessagingService.DbPublishResponse(new MergeDeliusReturnMessage()); + }, TDbQueue.MergeDelius); - dbMessagingService.DbLongSubscribe(async (message) => - { - await dbInteractionService.ClearDeliusStaging(); - dbMessagingService.DbPublishResponse(new ResultClearDeliusStaging()); - }, TDbQueue.ClearDeliusStaging); + dbMessagingService.SubscribeToDbRequest(async (message) => + { + await dbInteractionService.ClearDeliusStaging(); + dbMessagingService.DbPublishResponse(new ResultClearDeliusStaging()); + }, TDbQueue.ClearDeliusStaging); - dbMessagingService.DbLongSubscribe(async (message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { await dbInteractionService.StageOffloc(message.fileName); dbMessagingService.DbPublishResponse(new StageOfflocReturnMessage()); - mergingService.MergingPublish(new OfflocFilesCleanupMessage(message.fileName)); + mergingService.MergingPublish(new OfflocFilesCleanupMessage(message.fileName)); }, TDbQueue.StageOffloc); - dbMessagingService.DbLongSubscribe(async (message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { - await dbInteractionService.MergeOfflocPicture(); + await dbInteractionService.MergeOfflocPicture(message.fileName); await dbInteractionService.ClearOfflocStaging(); dbMessagingService.DbPublishResponse(new MergeOfflocReturnMessage()); }, TDbQueue.MergeOffloc); - dbMessagingService.DbLongSubscribe(async (message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { await dbInteractionService.ClearOfflocStaging(); dbMessagingService.DbPublishResponse(new ResultClearOfflocStaging()); }, TDbQueue.ClearOfflocStaging); - dbMessagingService.DbLongSubscribe(async (message) => + dbMessagingService.SubscribeToDbRequest(async (message) => { int res = await dbInteractionService.DeliusGetFileIdLastFull(); dbMessagingService.DbPublishResponse(new ResultDeliusGetLastFullMessage(res)); }, TDbQueue.DeliusGetLastFullId); + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + await dbInteractionService.CreateOfflocProcessedFileEntry(message.fileName, message.fileId, message.archiveName); + dbMessagingService.DbPublishResponse(new ResultOfflocFileProcessingStarted()); + }, TDbQueue.OfflocFileProcessingStarted); + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + await dbInteractionService.CreateDeliusProcessedFileEntry(message.fileName, message.fileId); + dbMessagingService.DbPublishResponse(new ResultDeliusFileProcessingStarted()); + }, TDbQueue.DeliusFileProcessingStarted); + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + await dbInteractionService.AssociateOfflocFileWithArchive(message.fileName, message.archiveName); + dbMessagingService.DbPublishResponse(new ResultAssociateOfflocFileWithArchiveMessage()); + }, TDbQueue.AssociateOfflocFileWithArchive); + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + bool result = await dbInteractionService.IsDeliusReadyForProcessing(); + dbMessagingService.DbPublishResponse(new IsDeliusReadyForProcessingReturnMessage(result)); + }, TDbQueue.IsDeliusReadyForProcessing); + + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + bool result = await dbInteractionService.IsOfflocReadyForProcessing(); + dbMessagingService.DbPublishResponse(new IsOfflocReadyForProcessingReturnMessage(result)); + }, TDbQueue.IsOfflocReadyForProcessing); + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + string? result = await dbInteractionService.GetLastProcessedOfflocFileName(); + dbMessagingService.DbPublishResponse(new ResultGetLastProcessedOfflocFileMessage(result)); + }, TDbQueue.GetLastProcessedOfflocFile); + + dbMessagingService.SubscribeToDbRequest(async (message) => + { + string? result = await dbInteractionService.GetLastProcessedDeliusFileName(); + dbMessagingService.DbPublishResponse(new ResultGetLastProcessedDeliusFileMessage(result)); + }, TDbQueue.GetLastProcessedDeliusFile); + } } diff --git a/src/DbInteractions/DbInteractions.csproj b/src/DbInteractions/DbInteractions.csproj index f0193b5..907ff37 100644 --- a/src/DbInteractions/DbInteractions.csproj +++ b/src/DbInteractions/DbInteractions.csproj @@ -13,12 +13,10 @@ - - diff --git a/src/DbInteractions/Services/DbInteractionService.cs b/src/DbInteractions/Services/DbInteractionService.cs index addb796..4c6b73e 100644 --- a/src/DbInteractions/Services/DbInteractionService.cs +++ b/src/DbInteractions/Services/DbInteractionService.cs @@ -27,18 +27,7 @@ public DbInteractionService(IStatusMessagingService messageService, { this.statusService = messageService; this.serverConfig = serverConfig; - - //Sql Server runs in docker all of the time so the docker implementation - //of IFileLocations is always needed. - //check if SQL is runnig in container - if (config.GetValue("DOTNET_ENVIRONMENT") == "Development") - { - this.fileLocations = new DockerFileLocations(); - } - else - { - this.fileLocations = fileLocations; - } + this.fileLocations = fileLocations; deliusStagingConnString = connStrings.deliusStagingConnectionString; offlocStagingConnString = connStrings.offlocStagingConnectionString; @@ -150,7 +139,11 @@ public async Task GetProcessedOfflocFileNames() { await conn.OpenAsync(); - SqlCommand cmd = new SqlCommand("SELECT FileName FROM [OfflocRunningPicture].[ProcessedFiles];", conn); + SqlCommand cmd = new SqlCommand(@" + SELECT FileName as [Name] FROM [OfflocRunningPicture].[ProcessedFiles] + UNION + SELECT ArchiveFileName as [Name] FROM [OfflocRunningPicture].[ProcessedFiles] WHERE ArchiveFileName IS NOT NULL", conn); + var reader = await cmd.ExecuteReaderAsync(); List fileNames = new List(); @@ -259,7 +252,7 @@ public async Task StandardiseDeliusStaging() statusService.StatusPublish(new StatusUpdateMessage("Delius staging database standardisation complete.")); } //Calls merge and then on completion - public async Task MergeDeliusPicture() + public async Task MergeDeliusPicture(string fileName) { SqlConnection conn = new SqlConnection(deliusPictureConnString); @@ -271,6 +264,8 @@ public async Task MergeDeliusPicture() command.CommandType = CommandType.StoredProcedure; command.CommandTimeout = 1200; + command.Parameters.AddWithValue("@fileName", fileName); + try { await command.ExecuteNonQueryAsync(); @@ -309,7 +304,7 @@ public async Task ClearDeliusStaging() statusService.StatusPublish(new StatusUpdateMessage("Delius staging database cleared.")); } - public async Task MergeOfflocPicture() + public async Task MergeOfflocPicture(string fileName) { SqlConnection offlocConn = new SqlConnection(offlocPictureConnString); @@ -321,6 +316,8 @@ public async Task MergeOfflocPicture() command.CommandType = CommandType.StoredProcedure; command.CommandTimeout = 1200; + command.Parameters.AddWithValue("@fileName", fileName); + try { var res = await command.ExecuteNonQueryAsync(); @@ -358,4 +355,235 @@ public async Task ClearOfflocStaging() statusService.StatusPublish(new StatusUpdateMessage("Offloc staging database cleared.")); } + + public async Task CreateOfflocProcessedFileEntry(string fileName, int fileId, string? archiveName = null) + { + SqlConnection offlocConn = new(offlocPictureConnString); + + using (offlocConn) + { + await offlocConn.OpenAsync(); + + var command = new SqlCommand(@"INSERT INTO [OfflocRunningPicture].[ProcessedFiles] (FileName, FileId, ArchiveFileName, Status) VALUES (@fileName, @fileId, @archiveName, @status)", offlocConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + command.Parameters.AddWithValue("@fileName", fileName); + command.Parameters.AddWithValue("@fileId", fileId); + command.Parameters.AddWithValue("@archiveName", (object?)archiveName ?? DBNull.Value); + command.Parameters.AddWithValue("@status", "Processing"); + + try + { + var res = await command.ExecuteNonQueryAsync(); + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return; + } + } + } + + public async Task CreateDeliusProcessedFileEntry(string fileName, string fileId) + { + SqlConnection deliusConn = new(deliusPictureConnString); + + using (deliusConn) + { + await deliusConn.OpenAsync(); + var command = new SqlCommand(@"INSERT INTO [DeliusRunningPicture].[ProcessedFiles] (FileName, FileId, Status) VALUES (@fileName, @fileId, @status)", deliusConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + + command.Parameters.AddWithValue("@fileName", fileName); + command.Parameters.AddWithValue("@fileId", fileId); + command.Parameters.AddWithValue("@status", "Processing"); + + try + { + var res = await command.ExecuteNonQueryAsync(); + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return; + } + } + } + + public async Task AssociateOfflocFileWithArchive(string fileName, string archiveName) + { + SqlConnection offlocConn = new(offlocPictureConnString); + + using (offlocConn) + { + await offlocConn.OpenAsync(); + + var command = new SqlCommand(@" + UPDATE [OfflocRunningPicture].[ProcessedFiles] + SET ArchiveFileName = @archiveName + WHERE FileName = @fileName", offlocConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + + command.Parameters.AddWithValue("@fileName", fileName); + command.Parameters.AddWithValue("@archiveName", archiveName); + + try + { + var res = await command.ExecuteNonQueryAsync(); + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return; + } + } + } + + public async Task IsDeliusReadyForProcessing() + { + SqlConnection deliusConn = new(deliusPictureConnString); + + using (deliusConn) + { + await deliusConn.OpenAsync(); + + var command = new SqlCommand(@" + SELECT CAST(IIF( + NOT EXISTS ( + SELECT 1 + FROM DeliusRunningPicture.ProcessedFiles + WHERE Status <> 'Merged' + ), + 1, + 0 + ) AS BIT)", deliusConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + + try + { +#pragma warning disable CS8605 // Unboxing a possibly null value. + bool result = (bool)await command.ExecuteScalarAsync(); +#pragma warning restore CS8605 // Unboxing a possibly null value. + return result; + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return false; + } + } + } + + public async Task IsOfflocReadyForProcessing() + { + SqlConnection offlocConn = new(offlocPictureConnString); + + using (offlocConn) + { + await offlocConn.OpenAsync(); + + var command = new SqlCommand(@" + SELECT CAST(IIF( + NOT EXISTS ( + SELECT 1 + FROM OfflocRunningPicture.ProcessedFiles + WHERE Status <> 'Merged' + ), + 1, + 0 + ) AS BIT)", offlocConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + + try + { +#pragma warning disable CS8605 // Unboxing a possibly null value. + bool result = (bool)await command.ExecuteScalarAsync(); +#pragma warning restore CS8605 // Unboxing a possibly null value. + return result; + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return false; + } + } + } + + public async Task GetLastProcessedOfflocFileName() + { + SqlConnection offlocConn = new(offlocPictureConnString); + + using (offlocConn) + { + await offlocConn.OpenAsync(); + + var command = new SqlCommand(@" + SELECT TOP 1 FileName + FROM OfflocRunningPicture.ProcessedFiles + ORDER BY ValidFrom DESC", offlocConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + + try + { +#pragma warning disable CS8605 // Unboxing a possibly null value. + string? result = (string?)await command.ExecuteScalarAsync(); +#pragma warning restore CS8605 // Unboxing a possibly null value. + return result; + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return null; + } + } + } + + public async Task GetLastProcessedDeliusFileName() + { + SqlConnection deliusConn = new(deliusPictureConnString); + + using (deliusConn) + { + await deliusConn.OpenAsync(); + + var command = new SqlCommand(@" + SELECT TOP 1 FileName + FROM DeliusRunningPicture.ProcessedFiles + ORDER BY ValidFrom DESC", deliusConn) + { + CommandType = CommandType.Text, + CommandTimeout = 1200 + }; + + try + { +#pragma warning disable CS8605 // Unboxing a possibly null value. + string? result = (string?)await command.ExecuteScalarAsync(); +#pragma warning restore CS8605 // Unboxing a possibly null value. + return result; + } + catch (SqlException e) + { + statusService.StatusPublish(new StatusUpdateMessage(e.Message)); + return null; + } + } + } } diff --git a/src/DbInteractions/Services/IDbInteractionService.cs b/src/DbInteractions/Services/IDbInteractionService.cs index be5c5c5..b2efff2 100644 --- a/src/DbInteractions/Services/IDbInteractionService.cs +++ b/src/DbInteractions/Services/IDbInteractionService.cs @@ -12,9 +12,16 @@ public interface IDbInteractionService Task GetProcessedOfflocIds(); Task StageDelius(string fileName, string filePath); Task StageOffloc(string fileName); - Task MergeDeliusPicture(); - Task MergeOfflocPicture(); + Task MergeDeliusPicture(string fileName); + Task MergeOfflocPicture(string fileName); Task StandardiseDeliusStaging(); Task ClearDeliusStaging(); Task ClearOfflocStaging(); + Task CreateOfflocProcessedFileEntry(string fileName, int fileId, string? archiveName = null); + Task CreateDeliusProcessedFileEntry(string fileName, string fileId); + Task AssociateOfflocFileWithArchive(string fileName, string archiveName); + Task IsDeliusReadyForProcessing(); + Task IsOfflocReadyForProcessing(); + Task GetLastProcessedOfflocFileName(); + Task GetLastProcessedDeliusFileName(); } diff --git a/src/Delius.Parser/Core/PostParser.cs b/src/Delius.Parser/Core/PostParser.cs index f6d2944..21c4c22 100644 --- a/src/Delius.Parser/Core/PostParser.cs +++ b/src/Delius.Parser/Core/PostParser.cs @@ -53,15 +53,19 @@ public async Task PostParse(string outputFolderPath) await PostParseFile(linkedConfigs); DisposeStreams(); + + var oldFile = Path.Combine(outputFolderPath, $"{baseFileName}.txt"); + var newFile = Path.Combine(outputFolderPath, $"{baseFileName}_new.txt"); - if (File.Exists($"{outputFolderPath}/{baseFileName}_new.txt")) + if (File.Exists(newFile)) { - File.Replace($"{outputFolderPath}/{baseFileName}_new.txt", $"{outputFolderPath}/{baseFileName}.txt", null); + File.Delete(oldFile); + File.Move(newFile, oldFile); } else { //Deletes original file afterwards. - File.Delete($"{outputFolderPath}/{baseFileName}.txt"); + File.Delete(oldFile); } } } @@ -146,7 +150,9 @@ private void InitializeWriters(PostParsingConfiguration[] postParsingConfigs) postParsingConfigs[i].NewFileName += "_new"; } - newFileWriters[i] = new StreamWriter($"{outputFolderPath}/{postParsingConfigs[i].NewFileName}.txt"); + newFileWriters[i] = new StreamWriter($"{outputFolderPath}/{postParsingConfigs[i].NewFileName}.txt", append: false, encoding: Encoding.Unicode); + newFileWriters[i].NewLine = "\r\n"; + NewFileWriters.Add(postParsingConfigs[i].NewFileName, newFileWriters[i]); NewFilePreviousLines.Add(postParsingConfigs[i].NewFileName, new HashSet(2500)); } diff --git a/src/Delius.Parser/Core/TextFileProcessor.cs b/src/Delius.Parser/Core/TextFileProcessor.cs index d2052f3..c3308e9 100644 --- a/src/Delius.Parser/Core/TextFileProcessor.cs +++ b/src/Delius.Parser/Core/TextFileProcessor.cs @@ -15,8 +15,6 @@ public TextFileProcessor(IStagingMessagingService messageService, DeliusProcesso { this.messageService = messageService; this.deliusProcessor = deliusProcessor; - - //messageService.StagingSubscribe(async(message) => await Process(), TStagingQueue.DeliusParser); } public async Task Process(string fileToBeParsed, string outputDirectory) diff --git a/src/Delius.Parser/Delius.Parser.csproj b/src/Delius.Parser/Delius.Parser.csproj index f5de233..ac890f7 100644 --- a/src/Delius.Parser/Delius.Parser.csproj +++ b/src/Delius.Parser/Delius.Parser.csproj @@ -8,11 +8,7 @@ ..\..\.. bed042b3-1d10-47f8-9b97-b516f3ae5895 - - - - - + Always @@ -27,7 +23,6 @@ - diff --git a/src/Delius.Parser/DeliusParserBackgroundService.cs b/src/Delius.Parser/DeliusParserBackgroundService.cs index d40b61e..4db7687 100644 --- a/src/Delius.Parser/DeliusParserBackgroundService.cs +++ b/src/Delius.Parser/DeliusParserBackgroundService.cs @@ -1,4 +1,5 @@ using Delius.Parser.Services; +using EnvironmentSetup; using FileStorage; using Messaging.Interfaces; using Messaging.Messages.DbMessages.Receiving; @@ -7,77 +8,48 @@ using Messaging.Messages.StatusMessages; using Messaging.Queues; using Microsoft.Extensions.Hosting; -using System.Globalization; namespace Delius.Parser; -public class DeliusParserBackgroundService : BackgroundService +public class DeliusParserBackgroundService( + IStagingMessagingService messageService, + IDbMessagingService dbService, + IStatusMessagingService statusService, + IParsingStrategy parseService) : BackgroundService { - private readonly IStagingMessagingService messageService; - private readonly IDbMessagingService dbService; - private readonly IStatusMessagingService statusService; - private readonly IParsingStrategy parseService; - private readonly IFileLocations fileLocations; - - private CultureInfo cultureInfo = new CultureInfo("en-GB"); - - public DeliusParserBackgroundService(IStagingMessagingService messageService, - IDbMessagingService dbService, IStatusMessagingService statusService, - IParsingStrategy parseStrategy, IFileLocations fileLocations) - { - this.messageService = messageService; - this.statusService = statusService; - this.dbService = dbService; - parseService = parseStrategy; - this.fileLocations = fileLocations; - } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await Task.Run(() => { - messageService.StagingSubscribe(async (message) => await ParseFilesAsync(), TStagingQueue.DeliusParser); + messageService.StagingSubscribe(async (message) => await ParseFileAsync(message), TStagingQueue.DeliusParser); }, stoppingToken); } - private async Task ParseFilesAsync() + private async Task ParseFileAsync(DeliusDownloadFinishedMessage message) { - string[] files = await GetFilesAsync(); + var file = message.fileName; - if (files.Length == 0) + if (await HasAlreadyBeenProcessedAsync(file)) { - statusService.StatusPublish(new StatusUpdateMessage($"No files found in {fileLocations.deliusInput}")); - messageService.StagingPublish(new DeliusParserFinishedMessage("No Files", "No Path", true)); + statusService.StatusPublish(new StatusUpdateMessage($"File {file} has already been processed")); + messageService.StagingPublish(new DeliusParserFinishedMessage("File already processed", "No Path", true)); } else { - await parseService.ParseFiles(files); + await BeginProcessing(file, message.FileId); } } - - //Gets a list of files in the target directory. - private async Task GetFilesAsync() + + private async Task BeginProcessing(string fileName, string fileId) { - var files = Directory.GetFiles(fileLocations.deliusInput); - - for (int i = 0; i !res.Contains(f)) - .OrderBy(f => DateOnly.Parse($"{f[27..29]}/{f[25..27]}/{f[21..25]}", cultureInfo)) - .ToArray(); + var request = new DeliusFileProcessingStarted(fileName, fileId); + await dbService.SendDbRequestAndWaitForResponse(request); + await parseService.ParseFileAsync(fileName); } - private async Task GetAlreadyProcessedFiles() + private async Task HasAlreadyBeenProcessedAsync(string file) { - var res = await dbService.DbTransientSubscribe(new GetDeliusFilesMessage()); - - return res.fileNames; + var res = await dbService.SendDbRequestAndWaitForResponse(new GetDeliusFilesMessage()); + return res.fileNames.Contains(file); } } \ No newline at end of file diff --git a/src/Delius.Parser/Program.cs b/src/Delius.Parser/Program.cs index c69200f..214dc40 100644 --- a/src/Delius.Parser/Program.cs +++ b/src/Delius.Parser/Program.cs @@ -31,16 +31,9 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); - bool parallel = builder.Configuration.GetValue("Parallel"); - if (parallel) - { - builder.Services.AddSingleton(); - } - else - { - builder.Services.AddSingleton(); - } + builder.Services.AddSingleton(); + //Ordering here v. important. builder.Services.AddSingleton(); builder.Services.AddSingleton(); diff --git a/src/Delius.Parser/Services/IParsingStrategy.cs b/src/Delius.Parser/Services/IParsingStrategy.cs index d32c0d9..e5d6f14 100644 --- a/src/Delius.Parser/Services/IParsingStrategy.cs +++ b/src/Delius.Parser/Services/IParsingStrategy.cs @@ -3,5 +3,5 @@ namespace Delius.Parser.Services; public interface IParsingStrategy { - Task ParseFiles(string[] files); + Task ParseFileAsync(string file); } diff --git a/src/Delius.Parser/Services/ParallelParsingStrategy.cs b/src/Delius.Parser/Services/ParallelParsingStrategy.cs deleted file mode 100644 index 1b8f353..0000000 --- a/src/Delius.Parser/Services/ParallelParsingStrategy.cs +++ /dev/null @@ -1,23 +0,0 @@ - -using Delius.Parser.Core; -using FileStorage; -using Messaging.Interfaces; - -namespace Delius.Parser.Services; - -public class ParallelParsingStrategy : ParsingStrategyBase, IParsingStrategy -{ - public ParallelParsingStrategy(IFileProcessor fileProcessor, IStatusMessagingService statusService, - IStagingMessagingService stagingService, IFileLocations fileLocations) - : base(fileProcessor, statusService, stagingService, fileLocations) { } - - public async Task ParseFiles(string[] files) - { - await Task.CompletedTask; - - Parallel.For(0, files.Length, async (i, f) => - { - await ParseFile(files[i]); - }); - } -} diff --git a/src/Delius.Parser/Services/ParsingStrategyBase.cs b/src/Delius.Parser/Services/ParsingStrategyBase.cs index 979e383..afd0235 100644 --- a/src/Delius.Parser/Services/ParsingStrategyBase.cs +++ b/src/Delius.Parser/Services/ParsingStrategyBase.cs @@ -22,7 +22,7 @@ public ParsingStrategyBase(IFileProcessor fp, IStatusMessagingService statusServ this.fileLocations = fileLocations; } - public async Task ParseFile(string file) + public async Task ProcessFile(string file) { statusService.StatusPublish(new StatusUpdateMessage($"Delius parser started on file {file}.")); diff --git a/src/Delius.Parser/Services/SequentialParsingStrategy.cs b/src/Delius.Parser/Services/SequentialParsingStrategy.cs index 5c46d91..0ca9372 100644 --- a/src/Delius.Parser/Services/SequentialParsingStrategy.cs +++ b/src/Delius.Parser/Services/SequentialParsingStrategy.cs @@ -1,4 +1,5 @@  +using System.Diagnostics; using Delius.Parser.Core; using FileStorage; using Messaging.Interfaces; @@ -12,12 +13,5 @@ public SequentialParsingStrategy(IFileProcessor fileProcessor, IFileLocations fileLocations) : base(fileProcessor, statusMessagingService, stagingMessagingService, fileLocations) { } - public async Task ParseFiles(string[] files) - { - files = files.Order().ToArray(); - for(int i = 0; i ProcessFile(file); } diff --git a/src/Delius.Parser/appsettings.json b/src/Delius.Parser/appsettings.json index a29829f..888ff7d 100644 --- a/src/Delius.Parser/appsettings.json +++ b/src/Delius.Parser/appsettings.json @@ -1,5 +1,4 @@ { "Serilog:WriteTo:1:Args:path": "%USERPROFILE%/DMS/logs/Delius.Parser-.txt", - "EnvFilePath": "..\\", - "Parallel": false + "EnvFilePath": "..\\" } diff --git a/src/DeliusRunningPictureDb/Stored Procedures/MergeAll.sql b/src/DeliusRunningPictureDb/Stored Procedures/MergeAll.sql index 4df0dbb..1941272 100644 --- a/src/DeliusRunningPictureDb/Stored Procedures/MergeAll.sql +++ b/src/DeliusRunningPictureDb/Stored Procedures/MergeAll.sql @@ -1,4 +1,5 @@ CREATE PROCEDURE [DeliusRunningPicture].[MergeAll] + @fileName NVARCHAR(255) NULL AS BEGIN SET NOCOUNT ON; @@ -30,6 +31,13 @@ BEGIN --Non Temporal Table EXEC [DeliusRunningPicture].[MergeStandardisationReference] + IF(@fileName IS NOT NULL) + BEGIN + UPDATE [DeliusRunningPicture].[ProcessedFiles] + SET [Status] = 'Merged' + WHERE [FileName] = @fileName; + END; + COMMIT TRANSACTION; END TRY BEGIN CATCH diff --git a/src/DeliusRunningPictureDb/Tables/ProcessedFiles.sql b/src/DeliusRunningPictureDb/Tables/ProcessedFiles.sql index d832eed..367d9de 100644 --- a/src/DeliusRunningPictureDb/Tables/ProcessedFiles.sql +++ b/src/DeliusRunningPictureDb/Tables/ProcessedFiles.sql @@ -4,6 +4,7 @@ CREATE TABLE [DeliusRunningPicture].[ProcessedFiles] [FileId] INT NOT NULL, [ValidFrom] DATETIME2 (7) GENERATED ALWAYS AS ROW START DEFAULT (sysutcdatetime()) NOT NULL, [ValidTo] DATETIME2 (7) GENERATED ALWAYS AS ROW END DEFAULT (CONVERT([datetime2],'9999-12-31 23:59:59.9999999')) NOT NULL, + [Status] VARCHAR (10) NULL, CONSTRAINT [PK_ProcessedFiles] PRIMARY KEY ([FileId] ASC), PERIOD FOR SYSTEM_TIME ([ValidFrom], [ValidTo]) ) ON [PRIMARY] WITH (SYSTEM_VERSIONING = ON (HISTORY_TABLE = [DeliusTemporal].[ProcessedFiles])) diff --git a/src/DeliusStagingDb/Stored Procedures/StageDelius.sql b/src/DeliusStagingDb/Stored Procedures/StageDelius.sql index 8965719..df6f73e 100644 --- a/src/DeliusStagingDb/Stored Procedures/StageDelius.sql +++ b/src/DeliusStagingDb/Stored Procedures/StageDelius.sql @@ -38,7 +38,10 @@ AS Declare @retMessage varchar(500); EXEC [DeliusStaging].[StandardiseData] @retMessage; - INSERT INTO [DeliusRunningPictureDb].[DeliusRunningPicture].[ProcessedFiles] (FileName, FileId) VALUES (@processedFile, CAST(SUBSTRING(@processedFile, 12, 4) AS int)); + UPDATE [DeliusRunningPictureDb].[DeliusRunningPicture].[ProcessedFiles] + SET [Status] = 'Imported' + WHERE FileName = @processedFile; + COMMIT TRANSACTION; END TRY BEGIN CATCH diff --git a/src/EnvironmentSetup/ConfigureEnvironment.cs b/src/EnvironmentSetup/ConfigureEnvironment.cs index bcbcce9..865f6ca 100644 --- a/src/EnvironmentSetup/ConfigureEnvironment.cs +++ b/src/EnvironmentSetup/ConfigureEnvironment.cs @@ -3,7 +3,6 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Hosting.WindowsServices; using Microsoft.Extensions.Logging; using Serilog; @@ -52,22 +51,11 @@ public static class ServiceConfiguration { public static IServiceCollection ConfigureServices(this IServiceCollection services, ConfigurationManager configManager) { - if (configManager.GetValue("RUNNING_IN_CONTAINER")) - { - //Path splitter registration goes here. - services.AddSingleton(); - } - else - { - services.AddSingleton( - f => new LocalFileLocations(configManager.GetValue("DMSFilesBasePath")!) - ); - } + services.AddSingleton( + f => new FileLocations(configManager.GetValue("DMSFilesBasePath")!) + ); - if(WindowsServiceHelpers.IsWindowsService()) - { - services.AddWindowsService(); - } + services.AddWindowsService(); services.ConfigureRabbit(configManager); services.ConfigureLogging(configManager); @@ -103,7 +91,7 @@ private static IServiceCollection ConfigureLogging(this IServiceCollection servi { Log.Logger = new LoggerConfiguration() .ReadFrom.Configuration(config) - .WriteTo.File(@".\logs\fatal.txt", Serilog.Events.LogEventLevel.Fatal) + .WriteTo.File(Path.Combine("logs", "fatal.txt"), Serilog.Events.LogEventLevel.Fatal) .CreateLogger(); var loggerFactory = LoggerFactory.Create(builder => diff --git a/src/EnvironmentSetup/EnvironmentSetup.csproj b/src/EnvironmentSetup/EnvironmentSetup.csproj index fcabc6b..12a1f3e 100644 --- a/src/EnvironmentSetup/EnvironmentSetup.csproj +++ b/src/EnvironmentSetup/EnvironmentSetup.csproj @@ -9,15 +9,10 @@ - - - - - + + + + + - - - - - diff --git a/src/EnvironmentSetup/FileLocations.cs b/src/EnvironmentSetup/FileLocations.cs new file mode 100644 index 0000000..6ff2225 --- /dev/null +++ b/src/EnvironmentSetup/FileLocations.cs @@ -0,0 +1,31 @@ + +namespace FileStorage; + + +public interface IFileLocations +{ + string basePath { get;} + string deliusInput { get; } + string deliusOutput { get; } + string offlocInput { get; } + string offlocOutput { get; } +} + +public class FileLocations : IFileLocations +{ + private string _basePath; + + public FileLocations(string basePath) + { + this._basePath = basePath.Replace("~", Environment.GetFolderPath(Environment.SpecialFolder.UserProfile)); + } + + public string basePath { get => _basePath; } + + public string deliusInput => Path.Combine(basePath, "Delius", "Input"); + public string deliusOutput => Path.Combine(basePath, "Delius", "Output"); + + //The offloc cleaner and offloc parser use the same input name. + public string offlocInput => Path.Combine(basePath, "Offloc", "Input"); + public string offlocOutput => Path.Combine(basePath, "Offloc", "Output"); +} \ No newline at end of file diff --git a/src/EnvironmentSetup/FilePatterns.cs b/src/EnvironmentSetup/FilePatterns.cs new file mode 100644 index 0000000..5d346ed --- /dev/null +++ b/src/EnvironmentSetup/FilePatterns.cs @@ -0,0 +1,3 @@ +namespace EnvironmentSetup; + +public record FilePatterns(string Delius, string Offloc); \ No newline at end of file diff --git a/src/EnvironmentSetup/PathSplitter.cs b/src/EnvironmentSetup/PathSplitter.cs deleted file mode 100644 index 396eb68..0000000 --- a/src/EnvironmentSetup/PathSplitter.cs +++ /dev/null @@ -1,9 +0,0 @@ - -namespace EnvironmentSetup; - -//Registered as a dependency so you don't have to split by both / and \ when -//dealing with paths. -public class PathSplitter -{ - public char splittingCharacter; -} diff --git a/src/FakeDataSeeder/FakeDataSeeder.csproj b/src/FakeDataSeeder/FakeDataSeeder.csproj index 4c6267a..e1382c0 100644 --- a/src/FakeDataSeeder/FakeDataSeeder.csproj +++ b/src/FakeDataSeeder/FakeDataSeeder.csproj @@ -2,7 +2,6 @@ Exe - net10.0 enable enable diff --git a/src/FileStorage/AWSService.cs b/src/FileStorage/AWSService.cs deleted file mode 100644 index 7948645..0000000 --- a/src/FileStorage/AWSService.cs +++ /dev/null @@ -1,21 +0,0 @@ - -namespace FileStorage; - -public class AWSService : IFileService -{ - - public AWSService() - { - - } - - public Task GetOfflocFile() - { - throw new NotImplementedException(); - } - - public Task GetOfflocFileNames() - { - throw new NotImplementedException(); - } -} diff --git a/src/FileStorage/DockerFileLocations.cs b/src/FileStorage/DockerFileLocations.cs deleted file mode 100644 index 3baee3d..0000000 --- a/src/FileStorage/DockerFileLocations.cs +++ /dev/null @@ -1,58 +0,0 @@ - -namespace FileStorage; - - -public interface IFileLocations -{ - string basePath { get;} - string deliusInput { get; } - string deliusOutput { get; } - string offlocInput { get; } - string offlocOutput { get; } - string keyLocations { get; } -} - -//TO-DO Put these classes in configurations in future. -//File paths for access from containers -public class DockerFileLocations : IFileLocations -{ - public DockerFileLocations() - { - this._basePath = "/app/"; - } - private string _basePath; - public string basePath { get => _basePath; } - - public string deliusInput => $"{basePath}delius/input"; - public string deliusOutput => $"{basePath}delius/output"; - - //The offloc cleaner and offloc parser use the same input name. - public string offlocInput => $"{basePath}offloc/input"; - public string offlocOutput => $"{basePath}offloc/output"; - public string keyLocations => $"{basePath}keys"; - - -} - -//Assumes dev work is always done on windows. -public class LocalFileLocations : IFileLocations -{ - //Default. - //public new string basePath = $"C:/Users/{System.Security.Principal.WindowsIdentity.GetCurrent().Name!.Split("\\").Last().ToLower()}/DMS/"; - private string _basePath; - public LocalFileLocations(string basePath) - { - this._basePath = basePath.Replace("~", Environment.GetFolderPath(Environment.SpecialFolder.UserProfile)); - } - public string basePath { get => _basePath; } - - public string deliusInput => $"{basePath}Delius/Input"; - public string deliusOutput => $"{basePath}Delius/Output"; - - //The offloc cleaner and offloc parser use the same input name. - public string offlocInput => $"{basePath}Offloc/Input"; - public string offlocOutput => $"{basePath}Offloc/Output"; - public string keyLocations => "~/AppData/Local/ASP.NET/DataProtection-Keys"; - - -} \ No newline at end of file diff --git a/src/FileStorage/FileExtensions.cs b/src/FileStorage/FileExtensions.cs deleted file mode 100644 index 5d5013d..0000000 --- a/src/FileStorage/FileExtensions.cs +++ /dev/null @@ -1,7 +0,0 @@ - -namespace FileStorage; - -public class FileExtensions -{ - public readonly string[] deliusFileExtensions = new string[] { "_new", }; -} diff --git a/src/FileStorage/FileStorage.csproj b/src/FileStorage/FileStorage.csproj deleted file mode 100644 index 7732feb..0000000 --- a/src/FileStorage/FileStorage.csproj +++ /dev/null @@ -1,13 +0,0 @@ - - - - enable - enable - - - - - - - - diff --git a/src/FileStorage/IFileService.cs b/src/FileStorage/IFileService.cs deleted file mode 100644 index a972401..0000000 --- a/src/FileStorage/IFileService.cs +++ /dev/null @@ -1,9 +0,0 @@ - -namespace FileStorage; - -public interface IFileService -{ - Task GetOfflocFileNames(); - Task GetOfflocFile(); - -} diff --git a/src/FileSync/Configuration/SourceOptions.cs b/src/FileSync/Configuration/SourceOptions.cs new file mode 100644 index 0000000..699c891 --- /dev/null +++ b/src/FileSync/Configuration/SourceOptions.cs @@ -0,0 +1,3 @@ +namespace FileSync.Configuration; + +public record FileSourceOptions(string Source); \ No newline at end of file diff --git a/src/FileSync/Configuration/SyncOptions.cs b/src/FileSync/Configuration/SyncOptions.cs new file mode 100644 index 0000000..08cf935 --- /dev/null +++ b/src/FileSync/Configuration/SyncOptions.cs @@ -0,0 +1,57 @@ +namespace FileSync.Configuration; + +public class SyncOptions +{ + /// + /// Indicates whether the application should process files and publish a message on startup. + /// + /// + /// When set to true, the application performs an initial file sync and publishes + /// a message to notify the downstream system that files are ready for processing. + /// + public required bool ProcessOnStartup { get; init; } + + /// + /// Indicates whether the application should process files and publish a message + /// after receiving confirmation that the previous batch completed successfully. + /// + /// + /// Triggered by . + /// + public required bool ProcessOnCompletion { get; init; } + + /// + /// Indicates whether the application should process files at a fixed time interval. + /// + /// + /// When enabled, the application will repeatedly process files based on the value + /// of . + /// + public bool ProcessOnTimer { get; set; } + + /// + /// The time interval, in seconds, between periodic processing runs. + /// + /// + /// This value is only used when is set to true. + /// It does not affect startup or completion-based processing, which are controlled + /// separately by and . + /// + /// + /// An integer representing the number of seconds between each automatic processing cycle. + /// A value less than or equal to zero disables interval-based processing. + /// + public required int ProcessTimerIntervalSeconds { get; init; } + + /// + /// Indicates whether the application should allow processing of files with timestamps + /// older than previously processed files. + /// + /// + /// When set to false, the application will throw an exception if it encounters a file + /// with a timestamp older than the most recently processed file, preventing potential + /// data inconsistencies. + /// When set to true, files will be processed regardless of their timestamp relative to previous runs. + /// + public bool AllowProcessingOlderFiles { get; init; } +} \ No newline at end of file diff --git a/src/FileSync/Core/FileSource.cs b/src/FileSync/Core/FileSource.cs new file mode 100644 index 0000000..eb0d636 --- /dev/null +++ b/src/FileSync/Core/FileSource.cs @@ -0,0 +1,16 @@ +namespace FileSync.Core; + +public abstract class FileSource +{ + /// + /// Returns a collection of Offloc files (including their paths and extensions) from the file source. + /// + public abstract Task> ListOfflocFilesAsync(CancellationToken cancellationToken = default); + + /// + /// Returns a collection of Delius files (including their paths and extensions) from the file source. + /// + public abstract Task> ListDeliusFilesAsync(CancellationToken cancellationToken = default); + + public abstract Task RetrieveFileAsync(string source, string target, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/FileSync/Extensions/FileExtensions.cs b/src/FileSync/Extensions/FileExtensions.cs new file mode 100644 index 0000000..848baa7 --- /dev/null +++ b/src/FileSync/Extensions/FileExtensions.cs @@ -0,0 +1,46 @@ +using System.Globalization; + +namespace FileSync.Extensions; + +public static class FileExtensions +{ + public static string GetFileId(this DeliusFile file) + { + var parts = file.Name.Split('_', StringSplitOptions.RemoveEmptyEntries); + return parts[1]; + } + + public static int? GetFileId(this OfflocFile file) + { + if (file.IsArchive) + { + return null; + } + + var parts = file.Name.Split('_', StringSplitOptions.RemoveEmptyEntries); + return int.Parse(parts[3]); + } + + public static DateOnly GetDatestamp(this DeliusFile file) + { + var parts = file.Name.Split('_', StringSplitOptions.RemoveEmptyEntries); + var datePart = parts.Last().Substring(0, 8); + return DateOnly.ParseExact(datePart, "yyyyMMdd", CultureInfo.InvariantCulture); + } + + public static DateOnly GetDatestamp(this OfflocFile file) + { + if (file.IsArchive) + { + var datePart = Path.GetFileNameWithoutExtension(file.Name); + return DateOnly.ParseExact(datePart, "yyyyMMdd", CultureInfo.InvariantCulture); + } + else + { + var id = file.GetFileId(); + var datePart = id.ToString()!.PadLeft(8, '0'); // 01/01/2024 becomes 1/01/2025, add leading zero to mitigate + return DateOnly.ParseExact(datePart, "ddMMyyyy", CultureInfo.InvariantCulture); + } + } + +} \ No newline at end of file diff --git a/src/FileSync/FileConstants.cs b/src/FileSync/FileConstants.cs new file mode 100644 index 0000000..f4d90f1 --- /dev/null +++ b/src/FileSync/FileConstants.cs @@ -0,0 +1,7 @@ +public static class FileConstants +{ + public const string DeliusFilePattern = @"^cfoextract_\d{1,5}_(full|diff)_\d{14}\.txt$"; + public const string OfflocFilePattern = @"^C_NOMIS_OFFENDER_\d{8}_.+\.dat$"; + public const string OfflocArchivePattern = @"^\d{8}\.zip$"; + public const string OfflocFileOrArchivePattern = @"^(?:\d{8}\.zip|C_NOMIS_OFFENDER_\d{8}_.+\.dat)$"; +} \ No newline at end of file diff --git a/src/FileSync/FileSync.csproj b/src/FileSync/FileSync.csproj new file mode 100644 index 0000000..4741c84 --- /dev/null +++ b/src/FileSync/FileSync.csproj @@ -0,0 +1,28 @@ + + + + Exe + enable + enable + 8317f847-53e8-4625-a75d-5aab845b6f17 + + + + + Always + + + + + + + + + + + + + + + + diff --git a/src/FileSync/FileSyncBackgroundService.cs b/src/FileSync/FileSyncBackgroundService.cs new file mode 100644 index 0000000..c83541b --- /dev/null +++ b/src/FileSync/FileSyncBackgroundService.cs @@ -0,0 +1,356 @@ +using System.Globalization; +using System.IO.Compression; +using System.Text.RegularExpressions; +using FileStorage; +using FileSync.Extensions; +using Messaging.Messages; +using Messaging.Messages.DbMessages.Receiving; +using Messaging.Messages.DbMessages.Sending; +using Messaging.Messages.MatchingMessages.Clustering; +using Messaging.Messages.StagingMessages; +using Messaging.Messages.StagingMessages.Delius; +using Messaging.Messages.StagingMessages.Offloc; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace FileSync; + +public class FileSyncBackgroundService( + ILogger logger, + IMatchingMessagingService matchingMessagingService, + IDbMessagingService dbMessagingService, + IStagingMessagingService stagingMessagingService, + IStatusMessagingService statusMessagingService, + IFileLocations fileLocations, + FileSource fileSource, + IOptions syncOptions) : BackgroundService, IDisposable +{ + private Timer? timer = null; + + private readonly SemaphoreSlim _lock = new(1, 1); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + logger.LogInformation("Starting service..."); + + try + { + if (syncOptions.Value.ProcessOnCompletion) + { + matchingMessagingService.MatchingSubscribe( + async msg => + { + await ProcessMessageAsync(msg, stoppingToken); + }, TMatchingQueue.ClusteringPostProcessingFinished); + } + + if (syncOptions.Value is { ProcessOnTimer: true, ProcessTimerIntervalSeconds: > 0 }) + { + timer = new Timer( + callback: async (state) => + { + logger.LogInformation("Timer elapsed, begin processing..."); + await ProcessAsync(stoppingToken); + }, + state: null, + dueTime: TimeSpan.FromSeconds(syncOptions.Value.ProcessTimerIntervalSeconds), + period: TimeSpan.FromSeconds(syncOptions.Value.ProcessTimerIntervalSeconds)); + } + + if (syncOptions.Value.ProcessOnStartup) + { + logger.LogInformation("Processing on startup configured, beginning processing..."); + await ProcessAsync(stoppingToken); + } + + } + catch (Exception ex) + { + logger.LogError(ex, ex.Message); + } + } + + async Task ProcessMessageAsync(Message message, CancellationToken cancellationToken) + { + logger.LogInformation($"{message.GetType().Name} received."); + await ProcessAsync(cancellationToken); + } + + async Task ProcessAsync(CancellationToken cancellationToken = default) + { + if (await _lock.WaitAsync(0, cancellationToken) is false) + { + logger.LogInformation("Previous processing still ongoing. Skipping this run."); + return; + } + + try + { + logger.LogInformation("Processing..."); + + var isDeliusReady = await IsDeliusReady(); + var isOfflocReady = await IsOfflocReady(); + + var notReady = (isDeliusReady, isOfflocReady) switch + { + (false, false) => "Delius and Offloc are not ready for processing.", + (false, true) => "Delius is not ready for processing.", + (true, false) => "Offloc is not ready for processing.", + _ => null + }; + + if (!string.IsNullOrEmpty(notReady)) + { + logger.LogWarning($"{notReady} Exiting."); + return; + } + + await PreKickoffTasks(); + + var unprocessedDeliusFile = await GetNextUnprocessedDeliusFileAsync(cancellationToken); + var unprocessedOfflocFile = await GetNextUnprocessedOfflocFileAsync(cancellationToken); + + // No files to process + if (unprocessedDeliusFile is null || unprocessedOfflocFile is null) + { + var notAvailable = (unprocessedDeliusFile, unprocessedOfflocFile) switch + { + (null, not null) => $"Delius file is not available, but Offloc file ({unprocessedOfflocFile.Name}) is ready.", + (not null, null) => $"Offloc file is not available, but Delius file ({unprocessedDeliusFile.Name}) is ready.", + _ => null + }; + + if (!string.IsNullOrEmpty(notAvailable)) + { + logger.LogWarning($"{notAvailable} Exiting."); + } + + return; + } + + // Check to see if a file newer than the unprocessed files have already been processed. + if (syncOptions.Value.AllowProcessingOlderFiles is false) + { + var isDeliusFileNewerThanLastProcessed = await IsDeliusFileNewerThanLastProcessed(unprocessedDeliusFile, cancellationToken); + var isOfflocFileNewerThanLastProcessed = await IsOfflocFileNewerThanLastProcessed(unprocessedOfflocFile, cancellationToken); + + var isEitherFileOutdated = (isDeliusFileNewerThanLastProcessed, isOfflocFileNewerThanLastProcessed) switch + { + (false, true) => "Delius file is older than the last processed Delius file.", + (true, false) => "Offloc file is older than the last processed Offloc file.", + (false, false) => "Both Delius and Offloc files are older than their last processed files.", + _ => null + }; + + if (!string.IsNullOrEmpty(isEitherFileOutdated)) + { + logger.LogError($"{isEitherFileOutdated} Exiting."); + return; + } + } + + logger.LogInformation($"Targeting: Delius file ({unprocessedDeliusFile.Name}), Offloc file ({unprocessedOfflocFile.Name})"); + + stagingMessagingService.StagingPublish(new DeliusDownloadFinishedMessage(unprocessedDeliusFile.Name, unprocessedDeliusFile.GetFileId())); + stagingMessagingService.StagingPublish(new OfflocDownloadFinished(unprocessedOfflocFile.Name, unprocessedOfflocFile.GetFileId()!.Value, unprocessedOfflocFile.ParentArchiveName)); + } + finally + { + logger.LogInformation("Processing complete."); + _lock.Release(); + } + } + + private async Task IsOfflocReady() + { + var response = await dbMessagingService.SendDbRequestAndWaitForResponse(new()); + return response.isReady; + } + + private async Task IsDeliusReady() + { + var response = await dbMessagingService.SendDbRequestAndWaitForResponse(new()); + return response.isReady; + } + + private async Task GetNextUnprocessedOfflocFileAsync(CancellationToken cancellationToken = default) + { + // Get files from file store (s3 / minio / local file system) + logger.LogInformation($"Retrieving Offloc files from {fileSource.GetType().Name} file store..."); + var offlocFileStore = await fileSource.ListOfflocFilesAsync(cancellationToken); + logger.LogInformation($"Found {offlocFileStore.Count} Offloc file(s) in {fileSource.GetType().Name} file store."); + + // Get already processed files + logger.LogInformation("Retrieving processed Offloc files..."); + var response = await dbMessagingService.SendDbRequestAndWaitForResponse(new GetOfflocFilesMessage()); + logger.LogInformation($"Retrieved {response.offlocFiles.Length} processed Offloc file(s)."); + + // Find latest unprocessed file + var unprocessedOfflocFiles = offlocFileStore + .ExceptBy(response.offlocFiles, file => Path.GetFileName(file.Name)) // Excludes processed + .OrderBy(f => f.GetDatestamp()) + .ToList(); + + logger.LogInformation($"Found {unprocessedOfflocFiles.Count} unprocessed Offloc file(s)."); + + var unprocessedOfflocFile = unprocessedOfflocFiles.FirstOrDefault(); + + // Not files to process + if(unprocessedOfflocFile is null) + { + return null; + } + + // Download the file + var downloadedFile = await fileSource.RetrieveFileAsync(unprocessedOfflocFile.Path, fileLocations.offlocInput, cancellationToken); + + // If the downloaded file is not an archive, process it. + if(unprocessedOfflocFile.IsArchive is false) + { + logger.LogInformation("Target Offloc file: " + unprocessedOfflocFile.Name); + return unprocessedOfflocFile; + } + + logger.LogInformation("Target Offloc archive: " + unprocessedOfflocFile.Name); + + // If the downloaded file is an archive, extract it and check the contained file. + // We only support archives containing a single Offloc (.dat) file. + logger.LogInformation("Extracting Offloc archive: " + unprocessedOfflocFile.Name); + await ZipFile.ExtractToDirectoryAsync(downloadedFile, fileLocations.offlocInput, overwriteFiles: true, cancellationToken); + File.Delete(downloadedFile); + + var filePath = Directory.GetFiles(fileLocations.offlocInput) + .Where(filePath => Regex.IsMatch(Path.GetFileName(filePath), FileConstants.OfflocFilePattern)) + .Single(); + + var file = Path.GetFileName(filePath); + + logger.LogInformation("Extracted Offloc file: " + file); + + // We have already processed the contents of the archive, get the next unprocessed file. + if(response.offlocFiles.Contains(file)) + { + // Associate the zip with the file name + logger.LogWarning("Already processed contents of archive: " + file); + await dbMessagingService.SendDbRequestAndWaitForResponse(new AssociateOfflocFileWithArchiveMessage(file, Path.GetFileName(downloadedFile))); + + // Remove the extracted file from the input directory - we have already processed it! + File.Delete(filePath); + + return await GetNextUnprocessedOfflocFileAsync(cancellationToken); + } + + return new OfflocFile(Path.Combine(fileLocations.offlocInput, file), Path.GetFileName(downloadedFile)); + } + + private async Task GetNextUnprocessedDeliusFileAsync(CancellationToken cancellationToken = default) + { + // Get files from file store (s3 / local file system) + var deliusFileStore = await fileSource.ListDeliusFilesAsync(cancellationToken); + + // Get already processed files + logger.LogInformation("Retrieving processed Delius files..."); + var response = await dbMessagingService.SendDbRequestAndWaitForResponse(new GetDeliusFilesMessage()); + logger.LogInformation($"Retrieved {response.fileNames.Length} processed Delius file(s)."); + + // Return unprocessed files + var unprocessedDeliusFiles = deliusFileStore + .ExceptBy(response.fileNames, file => Path.GetFileName(file.Name)) + .OrderBy(file => file.GetDatestamp()) + .ToList(); + + logger.LogInformation($"Found {unprocessedDeliusFiles.Count} unprocessed Delius file(s)."); + + var unprocessedDeliusFile = unprocessedDeliusFiles.FirstOrDefault(); + + if(unprocessedDeliusFile is null) + { + return null; + } + + var downloadedFile = await fileSource.RetrieveFileAsync(unprocessedDeliusFile.Path, fileLocations.deliusInput, cancellationToken); + + logger.LogInformation("Target Delius file: " + unprocessedDeliusFile.Name); + + return new DeliusFile(Path.GetFileName(downloadedFile)); + } + + private async Task PreKickoffTasks() + { + LogStatus("Publishing pre-kickoff messages..."); + stagingMessagingService.StagingPublish(new ClearHalfCleanedOfflocFiles()); + stagingMessagingService.StagingPublish(new ClearTemporaryDeliusFiles()); + + LogStatus("Pre-kickoff messages published. Beginning staging database tear down..."); + await dbMessagingService.SendDbRequestAndWaitForResponse(new ClearDeliusStaging()); + await dbMessagingService.SendDbRequestAndWaitForResponse(new ClearOfflocStaging()); + LogStatus("Staging database tear down complete."); + + // Delete any files in input directories + LogStatus("Clearing input directories..."); + int counter = 0; + + foreach (var file in Directory.GetFiles(fileLocations.deliusInput)) + { + File.Delete(file); + counter++; + } + + foreach (var file in Directory.GetFiles(fileLocations.offlocInput)) + { + File.Delete(file); + counter++; + } + + LogStatus("Input directories cleared. Deleted " + counter + " file(s)."); + } + + void LogStatus(string message) + { + logger.LogInformation(message); + statusMessagingService.StatusPublish(new StatusUpdateMessage(message)); + } + + public override void Dispose() + { + timer?.Dispose(); + } + + public async Task IsDeliusFileNewerThanLastProcessed(DeliusFile unprocessedDeliusFile, CancellationToken cancellationToken = default) + { + var lastProcessedDeliusFileName = await dbMessagingService.SendDbRequestAndWaitForResponse(new()); + + if(!string.IsNullOrEmpty(lastProcessedDeliusFileName.fileName)) + { + var file = new DeliusFile(lastProcessedDeliusFileName.fileName); + + if(file.GetDatestamp() >= unprocessedDeliusFile.GetDatestamp()) + { + logger.LogWarning($"The last processed Delius file ({file.Name}) is newer than or the same as the targeted Delius file ({unprocessedDeliusFile.Name})."); + return false; + } + } + + return true; + } + + public async Task IsOfflocFileNewerThanLastProcessed(OfflocFile unprocessedOfflocFile, CancellationToken cancellationToken = default) + { + var lastProcessedOfflocFileName = await dbMessagingService.SendDbRequestAndWaitForResponse(new()); + + if(!string.IsNullOrEmpty(lastProcessedOfflocFileName.fileName)) + { + var file = new OfflocFile(lastProcessedOfflocFileName.fileName); + + if(file.GetDatestamp() >= unprocessedOfflocFile.GetDatestamp()) + { + logger.LogWarning($"The last processed Offloc file ({file.Name}) is newer than or the same as the targeted Offloc file ({unprocessedOfflocFile.Name})."); + return false; + } + } + + return true; + } +} diff --git a/src/FileSync/GlobalUsings.cs b/src/FileSync/GlobalUsings.cs new file mode 100644 index 0000000..50330bb --- /dev/null +++ b/src/FileSync/GlobalUsings.cs @@ -0,0 +1,14 @@ +global using Amazon.S3; +global using EnvironmentSetup; +global using FileSync; +global using FileSync.Configuration; +global using FileSync.Core; +global using FileSync.Sources.Minio; +global using FileSync.Sources.Local; +global using FileSync.Sources.S3; +global using Messaging.Interfaces; +global using Messaging.Services; +global using Microsoft.Extensions.Configuration; +global using Microsoft.Extensions.DependencyInjection; +global using Microsoft.Extensions.Hosting; +global using Serilog; \ No newline at end of file diff --git a/src/FileSync/Model/DeliusFile.cs b/src/FileSync/Model/DeliusFile.cs new file mode 100644 index 0000000..2eaec30 --- /dev/null +++ b/src/FileSync/Model/DeliusFile.cs @@ -0,0 +1,4 @@ +public record DeliusFile(string Path) +{ + public string Name => System.IO.Path.GetFileName(Path); +} \ No newline at end of file diff --git a/src/FileSync/Model/OfflocFile.cs b/src/FileSync/Model/OfflocFile.cs new file mode 100644 index 0000000..e1bdd10 --- /dev/null +++ b/src/FileSync/Model/OfflocFile.cs @@ -0,0 +1,5 @@ +public record OfflocFile(string Path, string? ParentArchiveName = null) +{ + public string Name => System.IO.Path.GetFileName(Path); + public bool IsArchive => Path.EndsWith(".zip", StringComparison.OrdinalIgnoreCase); +} \ No newline at end of file diff --git a/src/FileSync/Program.cs b/src/FileSync/Program.cs new file mode 100644 index 0000000..3421d50 --- /dev/null +++ b/src/FileSync/Program.cs @@ -0,0 +1,64 @@ +Log.Logger = new LoggerConfiguration() + .Enrich.FromLogContext() + .WriteTo.Console() + .WriteTo.File(@"./logs/fatal.txt", Serilog.Events.LogEventLevel.Fatal) + .CreateBootstrapLogger(); + +try +{ + var builder = Host.CreateApplicationBuilder(args); + + builder.Configuration.ConfigureByEnvironment(); + + builder.Services.ConfigureServices(builder.Configuration); + + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + + builder.Services.AddOptions().BindConfiguration("SyncOptions"); + + var source = builder.Configuration.GetValue("FileSourceProvider")?.ToLowerInvariant(); + + if (source == "s3") + { + builder.Services.AddDefaultAWSOptions(builder.Configuration.GetAWSOptions()); + builder.Services.AddAWSService(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(sp => new FileSourceOptions(builder.Configuration.GetValue("AWS:S3:BucketName")!)); + } + else if (source == "filesystem") + { + builder.Services.AddSingleton(); + builder.Services.AddSingleton(sp => new FileSourceOptions(builder.Configuration.GetValue("FileSystem:SourceLocation")!)); + } + else if(source == "minio") + { + builder.AddMinioClient("minio"); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(sp => new FileSourceOptions(builder.Configuration.GetValue("MinIO:BucketName")!)); + } + else + { + throw new InvalidOperationException("Set 'FileSource' to 'S3', 'FileSystem', or 'MinIO' in configuration."); + } + + builder.Services.AddHostedService(); + + var app = builder.Build(); + Log.Information("Starting application"); + app.Run(); + + return 0; +} +catch (Exception ex) +{ + Log.Fatal(ex, "Host terminated unexpectedly"); + return 1; +} +finally +{ + Log.Information("Application stopping"); + await Log.CloseAndFlushAsync(); +} diff --git a/src/FileSync/Sources/Local/SystemFileSource.cs b/src/FileSync/Sources/Local/SystemFileSource.cs new file mode 100644 index 0000000..d99686e --- /dev/null +++ b/src/FileSync/Sources/Local/SystemFileSource.cs @@ -0,0 +1,52 @@ + +using System.Text.RegularExpressions; +using Microsoft.Extensions.Logging; + +namespace FileSync.Sources.Local; + +public class SystemFileSource( + ILogger logger, + FileSourceOptions options) : FileSource +{ + public override Task RetrieveFileAsync(string source, string target, CancellationToken cancellationToken = default) + { + logger.LogInformation($"Retrieving file from local system: {source}"); + + var sourcePath = Path.GetFullPath(source); + var targetPath = Path.GetFullPath(Path.Combine(target, Path.GetFileName(source))); + + if (sourcePath != targetPath) + { + File.Copy(sourcePath, targetPath); + } + + logger.LogInformation($"Retrieved file from local system: {targetPath}"); + + return Task.FromResult(targetPath); + } + + public override async Task> ListDeliusFilesAsync(CancellationToken cancellationToken = default) + { + logger.LogInformation($"Listing Delius files from local system at '{options.Source}' with pattern: {FileConstants.DeliusFilePattern}"); + var files = await GetFiles(options.Source, FileConstants.DeliusFilePattern); + return files.Select(file => new DeliusFile(file)).ToList(); + } + + public override async Task> ListOfflocFilesAsync(CancellationToken cancellationToken = default) + { + logger.LogInformation($"Listing Offloc files from local system at '{options.Source}' with pattern: {FileConstants.OfflocFileOrArchivePattern}"); + var files = await GetFiles(options.Source, FileConstants.OfflocFileOrArchivePattern); + return files.Select(file => new OfflocFile(file)).ToList(); + } + + private static Task> GetFiles(string location, string pattern) + { + IReadOnlyList files = Directory.GetFiles(location, "*", SearchOption.AllDirectories).Where(file => + { + var fileName = Path.GetFileName(file); + return Regex.IsMatch(fileName, pattern); + }).ToList(); + + return Task.FromResult(files); + } +} \ No newline at end of file diff --git a/src/FileSync/Sources/Minio/MinioFileSource.cs b/src/FileSync/Sources/Minio/MinioFileSource.cs new file mode 100644 index 0000000..d8b77ca --- /dev/null +++ b/src/FileSync/Sources/Minio/MinioFileSource.cs @@ -0,0 +1,103 @@ + +using System.Text.RegularExpressions; +using Microsoft.Extensions.Logging; +using Minio; +using Minio.DataModel.Args; + +namespace FileSync.Sources.Minio; + +public class MinioFileSource( + ILogger logger, + FileSourceOptions options, + IMinioClient minioClient) : FileSource +{ + private readonly SemaphoreSlim bucketInitLock = new(1, 1); + private bool bucketInitialised; + + public override async Task> ListDeliusFilesAsync(CancellationToken cancellationToken = default) + { + var files = await GetFilesAsync(options.Source, FileConstants.DeliusFilePattern, cancellationToken); + return files.Select(file => new DeliusFile(file)).ToList(); + } + + public override async Task> ListOfflocFilesAsync(CancellationToken cancellationToken = default) + { + var files = await GetFilesAsync(options.Source, FileConstants.OfflocFileOrArchivePattern, cancellationToken); + return files.Select(file => new OfflocFile(file)).ToList(); + } + private async Task> GetFilesAsync(string bucketName, string pattern, CancellationToken cancellationToken) + { + await EnsureBucketExistsAsync(bucketName, cancellationToken); + + var args = new ListObjectsArgs() + .WithBucket(bucketName); + + logger.LogInformation("Listing objects in Minio bucket '{Bucket}' matching pattern: {Pattern}", bucketName, pattern); + + return await minioClient.ListObjectsEnumAsync(args, cancellationToken) + .Where(obj => + { + var fileName = Path.GetFileName(obj.Key); + return Regex.IsMatch(fileName, pattern); + }) + .Select(obj => $"minio://{bucketName}/{obj.Key}") + .ToListAsync(cancellationToken); + } + + public override async Task RetrieveFileAsync(string source, string target, CancellationToken cancellationToken = default) + { + logger.LogInformation("Retrieving object from Minio: {Source}", source); + + var uri = new Uri(source); + var fileName = uri.AbsolutePath.TrimStart('/'); + var targetPath = Path.Combine(target, fileName); + + var args = new GetObjectArgs() + .WithBucket(uri.Host) + .WithObject(fileName) + .WithFile(targetPath); + + await minioClient.GetObjectAsync(args, cancellationToken); + + logger.LogInformation($"Retrieved object from Minio: {targetPath}"); + return targetPath; + } + + private async Task EnsureBucketExistsAsync(string bucketName, CancellationToken cancellationToken) + { + if (bucketInitialised) + { + return; + } + + await bucketInitLock.WaitAsync(cancellationToken); + try + { + if (bucketInitialised) + { + return; + } + + var exists = await minioClient.BucketExistsAsync( + new BucketExistsArgs().WithBucket(bucketName), + cancellationToken); + + if (!exists) + { + logger.LogInformation("Bucket '{Bucket}' does not exist in Minio. Creating it.", bucketName); + + await minioClient.MakeBucketAsync( + new MakeBucketArgs().WithBucket(bucketName), + cancellationToken); + + logger.LogInformation("Created bucket '{Bucket}' in Minio.", bucketName); + } + + bucketInitialised = true; + } + finally + { + bucketInitLock.Release(); + } + } +} \ No newline at end of file diff --git a/src/FileSync/Sources/S3/S3FileSource.cs b/src/FileSync/Sources/S3/S3FileSource.cs new file mode 100644 index 0000000..f3e4681 --- /dev/null +++ b/src/FileSync/Sources/S3/S3FileSource.cs @@ -0,0 +1,73 @@ +using System.Text.RegularExpressions; +using Amazon.S3.Model; +using Microsoft.Extensions.Logging; + +namespace FileSync.Sources.S3; + +public class S3FileSource( + ILogger logger, + FileSourceOptions options, + IAmazonS3 client) : FileSource +{ + public override async Task RetrieveFileAsync(string source, string target, CancellationToken cancellationToken = default) + { + logger.LogInformation($"Retrieving object from S3: {source}"); + + var uri = new Uri(source); + var fileName = Path.GetFileName(source); + var targetPath = Path.Combine(target, fileName); + + await client.DownloadToFilePathAsync( + bucketName: uri.Host, + objectKey: uri.LocalPath.TrimStart('/'), + filepath: targetPath, + new Dictionary { }, + cancellationToken); + + logger.LogInformation($"Retrieved object from S3: {targetPath}"); + + return targetPath; + } + + public override async Task> ListDeliusFilesAsync(CancellationToken cancellationToken = default) + { + logger.LogInformation($"Listing Delius files from S3 bucket '{options.Source}' with pattern: {FileConstants.DeliusFilePattern}"); + var files = await GetFilesAsync(options.Source, FileConstants.DeliusFilePattern, cancellationToken); + return files.Select(file => new DeliusFile(file)).ToList(); + } + + public override async Task> ListOfflocFilesAsync(CancellationToken cancellationToken = default) + { + logger.LogInformation($"Listing Offloc files from S3 bucket '{options.Source}' with pattern: {FileConstants.OfflocFileOrArchivePattern}"); + var files = await GetFilesAsync(options.Source, FileConstants.OfflocFileOrArchivePattern, cancellationToken); + return files.Select(file => new OfflocFile(file)).ToList(); + } + + private async Task> GetFilesAsync(string bucketName, string pattern, CancellationToken cancellationToken = default) + { + logger.LogInformation($"Attempting to list objects from S3 file store: {bucketName}"); + + var request = new ListObjectsV2Request + { + BucketName = bucketName + }; + + List files = []; + + var paginator = client.Paginators.ListObjectsV2(request); + + await foreach (var response in paginator.Responses.WithCancellation(cancellationToken)) + { + files.AddRange(response.S3Objects.Where(o => + { + var fileName = Path.GetFileName(o.Key); + return Regex.IsMatch(fileName, pattern); + }).Select(o => $"s3://{bucketName}/{o.Key}")); + } + + logger.LogInformation($"Retrieved {files.Count} object(s) from S3 file store: {bucketName}"); + + return files.AsReadOnly(); + } + +} \ No newline at end of file diff --git a/src/FileSync/appsettings.json b/src/FileSync/appsettings.json new file mode 100644 index 0000000..3ee8180 --- /dev/null +++ b/src/FileSync/appsettings.json @@ -0,0 +1,21 @@ +{ + "SyncOptions": { + "ProcessOnStartup": true, + "ProcessOnCompletion": true, + "ProcessOnTimer": true, + "ProcessTimerIntervalSeconds": 30, + "AllowProcessingOlderFiles": false + }, + "FileSourceProvider": "MinIO", // S3, FileSystem, MinIO + // Required if using 'S3' file source provider + "AWS": { + "Region": "eu-west-2", + "S3:BucketName": "" + }, + // Required if using 'FileSystem' file source provider + "FileSystem": { + "SourceLocation": "~/DMS/Files" + }, + // Required if using 'MinIO' file source provider + "MinIO:BucketName": "cfo-dms-files" +} \ No newline at end of file diff --git a/src/Import/Import.csproj b/src/Import/Import.csproj index 9497d02..c6af87b 100644 --- a/src/Import/Import.csproj +++ b/src/Import/Import.csproj @@ -16,7 +16,6 @@ - diff --git a/src/Import/ImportBackgroundService.cs b/src/Import/ImportBackgroundService.cs index 134f01c..6af2a6c 100644 --- a/src/Import/ImportBackgroundService.cs +++ b/src/Import/ImportBackgroundService.cs @@ -67,8 +67,8 @@ private async Task OnDeliusMessageReceived(DeliusParserFinishedMessage message) else { statusService.StatusPublish(new StatusUpdateMessage($"Importing Delius File.....")); - var res = await dbService.DbTransientSubscribe(new StageDeliusMessage(message.fileName, message.filePath)); - var msg = await dbService.DbTransientSubscribe(new MergeDeliusRunningPictureMessage()); + var res = await dbService.SendDbRequestAndWaitForResponse(new StageDeliusMessage(message.fileName, message.filePath)); + var msg = await dbService.SendDbRequestAndWaitForResponse(new MergeDeliusRunningPictureMessage(message.fileName)); } deliusSem.Release(); @@ -89,9 +89,8 @@ private async Task OnOfflocMessageReceived(OfflocParserFinishedMessage message) else { statusService.StatusPublish(new StatusUpdateMessage($"Importing Offloc File.....")); - var res = await dbService.DbTransientSubscribe(new StageOfflocMessage(message.filePath)); - var msg = await dbService.DbTransientSubscribe - (new MergeOfflocRunningPictureMessage()); + var res = await dbService.SendDbRequestAndWaitForResponse(new StageOfflocMessage(message.filePath)); + var msg = await dbService.SendDbRequestAndWaitForResponse(new MergeOfflocRunningPictureMessage(Path.GetFileName(message.filePath))); } offlocSem.Release(); offlocParserCompleted = true; diff --git a/src/Kickoff/KickoffService.cs b/src/Kickoff/KickoffService.cs index 8d2fb3e..ee58156 100644 --- a/src/Kickoff/KickoffService.cs +++ b/src/Kickoff/KickoffService.cs @@ -1,5 +1,4 @@ -using Kickoff.Options; -using Messaging.Interfaces; +using Messaging.Interfaces; using Messaging.Messages.DbMessages.Receiving; using Messaging.Messages.DbMessages.Sending; using Messaging.Messages.StagingMessages; @@ -16,7 +15,6 @@ public class KickoffService( IStagingMessagingService messageService, IStatusMessagingService statusService, IDbMessagingService dbService, - IOptions stagingOptions, IHostApplicationLifetime lifetime) : IHostedService { @@ -24,32 +22,18 @@ public async Task StartAsync(CancellationToken cancellationToken) { LogStatus("Performing pre-kickoff tasks..."); - await PreKickoffTasks(); + // await PreKickoffTasks(); - LogStatus("All pre-kickoff tasks complete. Begin staging message publish..."); + // LogStatus("All pre-kickoff tasks complete. Begin staging message publish..."); - var options = stagingOptions.Value; + // StagingMessage offlocStagingMessage = new OfflocDownloadFinished(); + // StagingMessage deliusStagingMessage = new DeliusDownloadFinishedMessage(); - StagingMessage offlocStagingMessage = new OfflocKickoffMessage(); - StagingMessage deliusStagingMessage = new DeliusKickoffMessage(); + // LogStatus($"Publishing {offlocStagingMessage.GetType().Name}..."); + // messageService.StagingPublish(offlocStagingMessage); - if(options.DownloadOfflocFiles is false) - { - LogStatus("Offloc file download is disabled, skipping download process."); - offlocStagingMessage = new OfflocDownloadFinished(); - } - - if (options.DownloadDeliusFiles is false) - { - LogStatus("Delius file download is disabled, skipping download process."); - deliusStagingMessage = new DeliusDownloadFinishedMessage(); - } - - LogStatus($"Publishing {offlocStagingMessage.GetType().Name}..."); - messageService.StagingPublish(offlocStagingMessage); - - LogStatus($"Publishing {deliusStagingMessage.GetType().Name}..."); - messageService.StagingPublish(deliusStagingMessage); + // LogStatus($"Publishing {deliusStagingMessage.GetType().Name}..."); + // messageService.StagingPublish(deliusStagingMessage); lifetime.StopApplication(); } @@ -69,8 +53,8 @@ private async Task PreKickoffTasks() messageService.StagingPublish(new ClearTemporaryDeliusFiles()); LogStatus("Pre-kickoff messages published. Beginning staging database tear down..."); - await dbService.DbTransientSubscribe(new ClearDeliusStaging()); - await dbService.DbTransientSubscribe(new ClearOfflocStaging()); + await dbService.SendDbRequestAndWaitForResponse(new ClearDeliusStaging()); + await dbService.SendDbRequestAndWaitForResponse(new ClearOfflocStaging()); LogStatus("Staging database tear down complete."); } diff --git a/src/Kickoff/Options/StagingOptions.cs b/src/Kickoff/Options/StagingOptions.cs deleted file mode 100644 index d96d608..0000000 --- a/src/Kickoff/Options/StagingOptions.cs +++ /dev/null @@ -1,7 +0,0 @@ -namespace Kickoff.Options; - -public class StagingOptions -{ - public bool DownloadOfflocFiles { get; set; } = false; - public bool DownloadDeliusFiles { get; set; } = false; -} diff --git a/src/Kickoff/Program.cs b/src/Kickoff/Program.cs index 5f1618f..1dd6554 100644 --- a/src/Kickoff/Program.cs +++ b/src/Kickoff/Program.cs @@ -3,7 +3,6 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.DependencyInjection; -using Kickoff.Options; using Messaging.Interfaces; using Messaging.Services; using Kickoff; @@ -23,8 +22,6 @@ builder.Services.ConfigureServices(builder.Configuration); - builder.Services.Configure(builder.Configuration.GetSection("Staging")); - builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); diff --git a/src/Logging/Logging.csproj b/src/Logging/Logging.csproj index f4b9ffa..0f79ae6 100644 --- a/src/Logging/Logging.csproj +++ b/src/Logging/Logging.csproj @@ -8,7 +8,6 @@ - diff --git a/src/Matching.Core/Matching.Core.csproj b/src/Matching.Core/Matching.Core.csproj index 9ca17d3..ca23e2c 100644 --- a/src/Matching.Core/Matching.Core.csproj +++ b/src/Matching.Core/Matching.Core.csproj @@ -10,7 +10,7 @@ enable enable - True + False diff --git a/src/Matching.Core/README.md b/src/Matching.Core/README.md new file mode 100644 index 0000000..b6a168f --- /dev/null +++ b/src/Matching.Core/README.md @@ -0,0 +1,82 @@ +# Matching.Core + +This library provides the core functionality for the matching engine. It defines the main interfaces and classes for implementing different matching strategies. + +## Core Components + +### `IMatcher` + +This is the primary interface for implementing a matcher. It has a single method: + +```csharp +Result Match(T source, T target); +``` + +- `T`: The type of the objects to compare. +- `Result`: A type that inherits from `MatcherResult`, representing the result of the match. + +### `IMatcher` + +This is a non-generic version of the `IMatcher` interface, intended for use when the types are not known at compile time. + +### `Matcher` + +This abstract class provides a base implementation of `IMatcher`. It handles type checking and casting, and provides an abstract `Match` method for subclasses to implement: + +```csharp +protected abstract Result Match(T? source, T? target); +``` + +### `MatcherResult` + +This abstract class represents the result of a matching operation. It contains the source and target objects and properties to indicate if one or both are missing. + +## Available Matchers + +The following matchers are available in the `Matching.Core.Matchers` namespace: + +- `CaverMatcher`: Matches based on the CAVER algorithm. +- `DateMatcher`: Matches dates with a configurable tolerance. +- `EqualityMatcher`: Performs a simple equality check. +- `JaroWinklerMatcher`: Uses the Jaro-Winkler distance algorithm for string comparison. +- `LevenshteinMatcher`: Uses the Levenshtein distance algorithm for string comparison. +- `PostcodeMatcher`: Matches UK postcodes. + +## Usage + +Here is an example of how to use the `EqualityMatcher`: + +```csharp +using Matching.Core.Matchers; + +var matcher = new EqualityMatcher(); +var result = matcher.Match("hello", "hello"); + +Console.WriteLine(result.Equal); // True +``` + +## Search + +### `Precedence` + +The `Precedence` class is a static class used to calculate a precedence score for a match. This score is used to rank potential matches. It takes in identifiers, last names, and dates of birth, and uses a combination of exact and fuzzy matching to generate a score. + +## Attributes + +### `MatcherAttribute` + +This attribute is used to decorate a matcher class with a key. This key can be used to dynamically discover and register matchers. + +```csharp +[Matcher("Equality")] +public class EqualityMatcher : Matcher +{ + // ... +} +``` + +## Utils + +The `Matching.Core.Utils` namespace contains various utility classes for string manipulation, date calculations, and implementations of the matching algorithms. + + diff --git a/src/Meow/Meow.csproj b/src/Meow/Meow.csproj index 72596b3..f03b6db 100644 --- a/src/Meow/Meow.csproj +++ b/src/Meow/Meow.csproj @@ -7,6 +7,8 @@ + + diff --git a/src/Messaging/Interfaces/IDbMessagingService.cs b/src/Messaging/Interfaces/IDbMessagingService.cs index 9b8aaf2..178214b 100644 --- a/src/Messaging/Interfaces/IDbMessagingService.cs +++ b/src/Messaging/Interfaces/IDbMessagingService.cs @@ -1,4 +1,4 @@ -using Messaging.Messages.DbMessages.Receiving; +using Messaging.Messages.DbMessages.Receiving; using Messaging.Messages.DbMessages.Sending; using Messaging.Queues; @@ -7,8 +7,8 @@ namespace Messaging.Interfaces; public interface IDbMessagingService { void DbPublishResponse(T message) where T : DbResponseMessage; - void DbLongSubscribe(Action handler, TDbQueue queue) where T : DbRequestMessage; - Task DbTransientSubscribe(T message) where T : DbRequestMessage where T2 : DbResponseMessage; + void SubscribeToDbRequest(Action handler, TDbQueue queue) where T : DbRequestMessage; + Task SendDbRequestAndWaitForResponse(T message) where T : DbRequestMessage where T2 : DbResponseMessage; //void DbUnsubscribe(Action handler, TDbQueue queue) where T : DbRequestMessage; } diff --git a/src/Messaging/Messages/DbMessages/Receiving/IsDeliusReadyForProcessingReturnMessage.cs b/src/Messaging/Messages/DbMessages/Receiving/IsDeliusReadyForProcessingReturnMessage.cs new file mode 100644 index 0000000..8674737 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/IsDeliusReadyForProcessingReturnMessage.cs @@ -0,0 +1,21 @@ +using Messaging.Messages.StatusMessages; +using Messaging.Queues; +using System.Text.Json.Serialization; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class IsDeliusReadyForProcessingReturnMessage : DbResponseMessage +{ + public bool isReady; + public override StatusUpdateMessage StatusMessage => new StatusUpdateMessage("Received Delius readiness: " + isReady); + + [JsonConstructor] + public IsDeliusReadyForProcessingReturnMessage() + { } + + public IsDeliusReadyForProcessingReturnMessage(bool isReady) + { + Queue = TDbQueue.IsDeliusReadyForProcessingResult; + this.isReady = isReady; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Receiving/IsOfflocReadyForProcessingReturnMessage.cs b/src/Messaging/Messages/DbMessages/Receiving/IsOfflocReadyForProcessingReturnMessage.cs new file mode 100644 index 0000000..ee03513 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/IsOfflocReadyForProcessingReturnMessage.cs @@ -0,0 +1,21 @@ +using Messaging.Messages.StatusMessages; +using Messaging.Queues; +using System.Text.Json.Serialization; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class IsOfflocReadyForProcessingReturnMessage : DbResponseMessage +{ + public bool isReady; + public override StatusUpdateMessage StatusMessage => new StatusUpdateMessage("Received Offloc readiness: " + isReady); + + [JsonConstructor] + public IsOfflocReadyForProcessingReturnMessage() + { } + + public IsOfflocReadyForProcessingReturnMessage(bool isReady) + { + Queue = TDbQueue.IsOfflocReadyForProcessingResult; + this.isReady = isReady; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Receiving/ResultAssociateOfflocFileWithArchiveMessage.cs b/src/Messaging/Messages/DbMessages/Receiving/ResultAssociateOfflocFileWithArchiveMessage.cs new file mode 100644 index 0000000..550dc1d --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/ResultAssociateOfflocFileWithArchiveMessage.cs @@ -0,0 +1,14 @@ +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class ResultAssociateOfflocFileWithArchiveMessage : DbResponseMessage +{ + public override StatusUpdateMessage StatusMessage => new("Offloc file associated with archive"); + + public ResultAssociateOfflocFileWithArchiveMessage() + { + Queue = TDbQueue.ResultAssociateOfflocFileWithArchive; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Receiving/ResultDeliusFileProcessingStarted.cs b/src/Messaging/Messages/DbMessages/Receiving/ResultDeliusFileProcessingStarted.cs new file mode 100644 index 0000000..80b541f --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/ResultDeliusFileProcessingStarted.cs @@ -0,0 +1,14 @@ +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class ResultDeliusFileProcessingStarted : DbResponseMessage +{ + public override StatusUpdateMessage StatusMessage => new("Delius file processing started"); + + public ResultDeliusFileProcessingStarted() + { + Queue = TDbQueue.ResultDeliusFileProcessingStarted; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Receiving/ResultGetLastProcessedDeliusFileMessage.cs b/src/Messaging/Messages/DbMessages/Receiving/ResultGetLastProcessedDeliusFileMessage.cs new file mode 100644 index 0000000..e200fc2 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/ResultGetLastProcessedDeliusFileMessage.cs @@ -0,0 +1,23 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class ResultGetLastProcessedDeliusFileMessage : DbResponseMessage +{ + public string? fileName; + + public override StatusUpdateMessage StatusMessage => new(); + + [JsonConstructor] + public ResultGetLastProcessedDeliusFileMessage() + { + Queue = TDbQueue.ResultLastProcessedDeliusFile; + } + + public ResultGetLastProcessedDeliusFileMessage(string? fileName) : this() + { + this.fileName = fileName; + } +} diff --git a/src/Messaging/Messages/DbMessages/Receiving/ResultGetLastProcessedOfflocFileMessage.cs b/src/Messaging/Messages/DbMessages/Receiving/ResultGetLastProcessedOfflocFileMessage.cs new file mode 100644 index 0000000..975ad1f --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/ResultGetLastProcessedOfflocFileMessage.cs @@ -0,0 +1,23 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class ResultGetLastProcessedOfflocFileMessage : DbResponseMessage +{ + public string? fileName; + + public override StatusUpdateMessage StatusMessage => new(); + + [JsonConstructor] + public ResultGetLastProcessedOfflocFileMessage() + { + Queue = TDbQueue.ResultLastProcessedOfflocFile; + } + + public ResultGetLastProcessedOfflocFileMessage(string? fileName) : this() + { + this.fileName = fileName; + } +} diff --git a/src/Messaging/Messages/DbMessages/Receiving/ResultOfflocFileProcessingStarted.cs b/src/Messaging/Messages/DbMessages/Receiving/ResultOfflocFileProcessingStarted.cs new file mode 100644 index 0000000..ca4c348 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Receiving/ResultOfflocFileProcessingStarted.cs @@ -0,0 +1,16 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Receiving; + +public class ResultOfflocFileProcessingStarted : DbResponseMessage +{ + public override StatusUpdateMessage StatusMessage => new("Offloc file processing started"); + + [JsonConstructor] + public ResultOfflocFileProcessingStarted() + { + Queue = TDbQueue.ResultOfflocFileProcessingStarted; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Sending/AssociateOfflocFileWithArchiveMessage.cs b/src/Messaging/Messages/DbMessages/Sending/AssociateOfflocFileWithArchiveMessage.cs new file mode 100644 index 0000000..1c7b9f0 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/AssociateOfflocFileWithArchiveMessage.cs @@ -0,0 +1,22 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class AssociateOfflocFileWithArchiveMessage : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new(); + + public string fileName { get; set; } + public string archiveName { get; set; } + + [JsonConstructor] + public AssociateOfflocFileWithArchiveMessage(string fileName, string archiveName) + { + Queue = TDbQueue.AssociateOfflocFileWithArchive; + ReplyQueue = TDbQueue.ResultAssociateOfflocFileWithArchive; + this.fileName = fileName; + this.archiveName = archiveName; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Sending/DeliusFileProcessingStarted.cs b/src/Messaging/Messages/DbMessages/Sending/DeliusFileProcessingStarted.cs new file mode 100644 index 0000000..0dcbd37 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/DeliusFileProcessingStarted.cs @@ -0,0 +1,26 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class DeliusFileProcessingStarted : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new(); + + public string fileName = string.Empty; + public string fileId = string.Empty; + + [JsonConstructor] + public DeliusFileProcessingStarted() + { + Queue = TDbQueue.DeliusFileProcessingStarted; + ReplyQueue = TDbQueue.ResultDeliusFileProcessingStarted; + } + + public DeliusFileProcessingStarted(string fileName, string fileId) : this() + { + this.fileName = fileName; + this.fileId = fileId; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Sending/GetLastProcessedDeliusFile.cs b/src/Messaging/Messages/DbMessages/Sending/GetLastProcessedDeliusFile.cs new file mode 100644 index 0000000..3781c47 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/GetLastProcessedDeliusFile.cs @@ -0,0 +1,15 @@ +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class GetLastProcessedDeliusFile : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new("Getting last processed delius file."); + + public GetLastProcessedDeliusFile() + { + Queue = TDbQueue.GetLastProcessedDeliusFile; + ReplyQueue = TDbQueue.ResultLastProcessedDeliusFile; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Sending/GetLastProcessedOfflocFile.cs b/src/Messaging/Messages/DbMessages/Sending/GetLastProcessedOfflocFile.cs new file mode 100644 index 0000000..8dfab26 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/GetLastProcessedOfflocFile.cs @@ -0,0 +1,16 @@ +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class GetLastProcessedOfflocFile : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new("Getting last processed offloc file."); + + public GetLastProcessedOfflocFile() + { + Queue = TDbQueue.GetLastProcessedOfflocFile; + ReplyQueue = TDbQueue.ResultLastProcessedOfflocFile; + } + +} \ No newline at end of file diff --git a/src/Messaging/Messages/DbMessages/Sending/IsDeliusReadyForProcessingMessage.cs b/src/Messaging/Messages/DbMessages/Sending/IsDeliusReadyForProcessingMessage.cs new file mode 100644 index 0000000..5fcd057 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/IsDeliusReadyForProcessingMessage.cs @@ -0,0 +1,17 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class IsDeliusReadyForProcessingMessage : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new("Sent request to check if delius is ready for processing."); + + [JsonConstructor] + public IsDeliusReadyForProcessingMessage() + { + Queue = TDbQueue.IsDeliusReadyForProcessing; + ReplyQueue = TDbQueue.IsDeliusReadyForProcessingResult; + } +} diff --git a/src/Messaging/Messages/DbMessages/Sending/IsOfflocReadyForProcessingMessage.cs b/src/Messaging/Messages/DbMessages/Sending/IsOfflocReadyForProcessingMessage.cs new file mode 100644 index 0000000..c3f938d --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/IsOfflocReadyForProcessingMessage.cs @@ -0,0 +1,17 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class IsOfflocReadyForProcessingMessage : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new("Sent request to check if offloc is ready for processing."); + + [JsonConstructor] + public IsOfflocReadyForProcessingMessage() + { + Queue = TDbQueue.IsOfflocReadyForProcessing; + ReplyQueue = TDbQueue.IsOfflocReadyForProcessingResult; + } +} diff --git a/src/Messaging/Messages/DbMessages/Sending/MergeDeliusRunningPictureMessage.cs b/src/Messaging/Messages/DbMessages/Sending/MergeDeliusRunningPictureMessage.cs index 4e8d883..a74395c 100644 --- a/src/Messaging/Messages/DbMessages/Sending/MergeDeliusRunningPictureMessage.cs +++ b/src/Messaging/Messages/DbMessages/Sending/MergeDeliusRunningPictureMessage.cs @@ -10,10 +10,13 @@ public class MergeDeliusRunningPictureMessage : DbRequestMessage public override StatusUpdateMessage StatusMessage => new StatusUpdateMessage("Merging into delius running picture started."); + public string fileName { get; set; } + [JsonConstructor] - public MergeDeliusRunningPictureMessage() + public MergeDeliusRunningPictureMessage(string fileName) { Queue = TDbQueue.MergeDelius; ReplyQueue = TDbQueue.ResultMergeDelius; + this.fileName = fileName; } } diff --git a/src/Messaging/Messages/DbMessages/Sending/MergeOfflocRunningPictureMessage.cs b/src/Messaging/Messages/DbMessages/Sending/MergeOfflocRunningPictureMessage.cs index 6018f4f..001b33e 100644 --- a/src/Messaging/Messages/DbMessages/Sending/MergeOfflocRunningPictureMessage.cs +++ b/src/Messaging/Messages/DbMessages/Sending/MergeOfflocRunningPictureMessage.cs @@ -10,10 +10,13 @@ public class MergeOfflocRunningPictureMessage : DbRequestMessage public override StatusUpdateMessage StatusMessage => new StatusUpdateMessage("Merging into Offloc running picture started."); + public string fileName { get; set; } + [JsonConstructor] - public MergeOfflocRunningPictureMessage() + public MergeOfflocRunningPictureMessage(string fileName) { Queue = TDbQueue.MergeOffloc; ReplyQueue = TDbQueue.ResultMergeOffloc; + this.fileName = fileName; } } diff --git a/src/Messaging/Messages/DbMessages/Sending/OfflocFileProcessingStarted.cs b/src/Messaging/Messages/DbMessages/Sending/OfflocFileProcessingStarted.cs new file mode 100644 index 0000000..fa59141 --- /dev/null +++ b/src/Messaging/Messages/DbMessages/Sending/OfflocFileProcessingStarted.cs @@ -0,0 +1,28 @@ +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; +using Messaging.Queues; + +namespace Messaging.Messages.DbMessages.Sending; + +public class OfflocFileProcessingStarted : DbRequestMessage +{ + public override StatusUpdateMessage StatusMessage => new(); + + public string fileName = string.Empty; + public int fileId; + public string? archiveName; + + [JsonConstructor] + public OfflocFileProcessingStarted() + { + Queue = TDbQueue.OfflocFileProcessingStarted; + ReplyQueue = TDbQueue.ResultOfflocFileProcessingStarted; + } + + public OfflocFileProcessingStarted(string fileName, int fileId, string? archiveName = null) : this() + { + this.fileName = fileName; + this.fileId = fileId; + this.archiveName = archiveName; + } +} \ No newline at end of file diff --git a/src/Messaging/Messages/StagingMessages/Delius/DeliusDownloadFinishedMessage.cs b/src/Messaging/Messages/StagingMessages/Delius/DeliusDownloadFinishedMessage.cs index 9594f2c..ad1512c 100644 --- a/src/Messaging/Messages/StagingMessages/Delius/DeliusDownloadFinishedMessage.cs +++ b/src/Messaging/Messages/StagingMessages/Delius/DeliusDownloadFinishedMessage.cs @@ -1,16 +1,21 @@ -using Messaging.Messages.StatusMessages; +using System.Text.Json.Serialization; +using Messaging.Messages.StatusMessages; using Messaging.Queues; namespace Messaging.Messages.StagingMessages; public class DeliusDownloadFinishedMessage : StagingMessage { - //Make file specific in future. - public override StatusUpdateMessage StatusMessage => - new StatusUpdateMessage(); + public override StatusUpdateMessage StatusMessage => new(); - public DeliusDownloadFinishedMessage() + public string FileId { get; set; } + + [JsonConstructor] + public DeliusDownloadFinishedMessage(string fileName, string fileId) { routingKey = TStagingQueue.DeliusParser; + base.fileName = fileName; + FileId = fileId; } + } diff --git a/src/Messaging/Messages/StagingMessages/Offloc/OfflocDownloadFinished.cs b/src/Messaging/Messages/StagingMessages/Offloc/OfflocDownloadFinished.cs index 64bcb85..3f691cf 100644 --- a/src/Messaging/Messages/StagingMessages/Offloc/OfflocDownloadFinished.cs +++ b/src/Messaging/Messages/StagingMessages/Offloc/OfflocDownloadFinished.cs @@ -7,12 +7,17 @@ namespace Messaging.Messages.StagingMessages; public class OfflocDownloadFinished : StagingMessage { - public override StatusUpdateMessage StatusMessage => - new StatusUpdateMessage(); + public override StatusUpdateMessage StatusMessage => new(); + + public string? ArchiveFileName { get; set; } + public int FileId { get; set; } [JsonConstructor] - public OfflocDownloadFinished() + public OfflocDownloadFinished(string fileName, int fileId, string? archiveFileName = null) { routingKey = TStagingQueue.OfflocCleaner; + base.fileName = fileName; + FileId = fileId; + ArchiveFileName = archiveFileName; } } diff --git a/src/Messaging/Messages/StagingMessages/Offloc/OfflocKickoffMessage.cs b/src/Messaging/Messages/StagingMessages/Offloc/OfflocKickoffMessage.cs deleted file mode 100644 index 2636c4a..0000000 --- a/src/Messaging/Messages/StagingMessages/Offloc/OfflocKickoffMessage.cs +++ /dev/null @@ -1,18 +0,0 @@ -using Messaging.Messages.StatusMessages; -using Messaging.Queues; -using System.Text.Json.Serialization; - -namespace Messaging.Messages.StagingMessages; - -public class OfflocKickoffMessage : StagingMessage -{ - public override StatusUpdateMessage StatusMessage => - new StatusUpdateMessage(); - - [JsonConstructor] - public OfflocKickoffMessage() - { - routingKey = TStagingQueue.OfflocFileDownload; - - } -} diff --git a/src/Messaging/Queues/TDbQueue.cs b/src/Messaging/Queues/TDbQueue.cs index 1e8fdc3..ccd511a 100644 --- a/src/Messaging/Queues/TDbQueue.cs +++ b/src/Messaging/Queues/TDbQueue.cs @@ -6,6 +6,7 @@ public enum TDbQueue //These queues will use Rabbit RPC. //Outgoing requests. + AssociateOfflocFileWithArchive, GetProcessedOfflocFiles, GetOfflocFileDates, DeliusGetLastFullId, @@ -17,6 +18,12 @@ public enum TDbQueue StandardiseDelius, ClearOfflocStaging, ClearDeliusStaging, + OfflocFileProcessingStarted, + DeliusFileProcessingStarted, + IsDeliusReadyForProcessing, + IsOfflocReadyForProcessing, + GetLastProcessedDeliusFile, + GetLastProcessedOfflocFile, //Incoming requests. ReturnedOfflocFiles, ReturnedOfflocFileDates, @@ -28,5 +35,12 @@ public enum TDbQueue ResultMergeDelius, ResultStandardiseDelius, ResultClearDelius, - ResultClearOffloc + ResultClearOffloc, + ResultOfflocFileProcessingStarted, + ResultDeliusFileProcessingStarted, + ResultAssociateOfflocFileWithArchive, + IsDeliusReadyForProcessingResult, + IsOfflocReadyForProcessingResult, + ResultLastProcessedDeliusFile, + ResultLastProcessedOfflocFile } diff --git a/src/Messaging/Services/RabbitService.cs b/src/Messaging/Services/RabbitService.cs index b7e3560..164f8a1 100644 --- a/src/Messaging/Services/RabbitService.cs +++ b/src/Messaging/Services/RabbitService.cs @@ -1,4 +1,4 @@ -using RabbitMQ.Client; +using RabbitMQ.Client; using System.Text; using RabbitMQ.Client.Events; using System.Text.Json; @@ -160,8 +160,7 @@ public void DbPublishResponse(T message) where T : DbResponseMessage AssociatedMessagePublish(message); } - //Subscriptions when waiting for requests are long-lived. - public void DbLongSubscribe(Action handler, TDbQueue queue) where T : DbRequestMessage + public void SubscribeToDbRequest(Action handler, TDbQueue queue) where T : DbRequestMessage { var consumer = new EventingBasicConsumer(channel); channel.BasicConsume(queue.ToString(), true, consumer); @@ -173,7 +172,7 @@ public void DbLongSubscribe(Action handler, TDbQueue queue) where T : DbRe }; } - public async Task DbTransientSubscribe(TRequest message) + public async Task SendDbRequestAndWaitForResponse(TRequest message) where TRequest : DbRequestMessage where TResponse : DbResponseMessage { diff --git a/src/Offloc.Cleaner/Offloc.Cleaner.csproj b/src/Offloc.Cleaner/Offloc.Cleaner.csproj index 8cedf8d..475a4c4 100644 --- a/src/Offloc.Cleaner/Offloc.Cleaner.csproj +++ b/src/Offloc.Cleaner/Offloc.Cleaner.csproj @@ -11,13 +11,10 @@ - - - diff --git a/src/Offloc.Cleaner/OfflocCleanerBackgroundService.cs b/src/Offloc.Cleaner/OfflocCleanerBackgroundService.cs index d1552a8..2125766 100644 --- a/src/Offloc.Cleaner/OfflocCleanerBackgroundService.cs +++ b/src/Offloc.Cleaner/OfflocCleanerBackgroundService.cs @@ -1,4 +1,6 @@ -using FileStorage; +using System.Globalization; +using EnvironmentSetup; +using FileStorage; using Messaging.Interfaces; using Messaging.Messages.DbMessages.Receiving; using Messaging.Messages.DbMessages.Sending; @@ -10,67 +12,39 @@ namespace Offloc.Cleaner; -public class OfflocCleanerBackgroundService : BackgroundService -{ - private readonly IStagingMessagingService stagingService; - private readonly IDbMessagingService dbService; - private readonly IStatusMessagingService statusService; - private readonly IFileLocations fileLocations; - private readonly ICleaningStrategy cleaningService; - - public OfflocCleanerBackgroundService(IStagingMessagingService stagingService, - IDbMessagingService dbService, ICleaningStrategy cleaningService, - IStatusMessagingService statusService, IFileLocations fileLocations) - { - this.stagingService = stagingService; - this.dbService = dbService; - this.statusService = statusService; - this.cleaningService = cleaningService; - this.fileLocations = fileLocations; - } - +public class OfflocCleanerBackgroundService( + IStagingMessagingService stagingService, + IDbMessagingService dbService, + ICleaningStrategy cleaningService, + IStatusMessagingService statusService) : BackgroundService +{ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - //stagingService.StagingSubscribe(async (message) => await ParseFile(), TStagingQueue.OfflocCleaner); await Task.Run(() => { - stagingService.StagingSubscribe(async (message) => await ParseFilesAsync(), TStagingQueue.OfflocCleaner); + stagingService.StagingSubscribe(async (message) => await ParseFileAsync(message), TStagingQueue.OfflocCleaner); }, stoppingToken); } - private async Task ParseFilesAsync() + private async Task ParseFileAsync(OfflocDownloadFinished message) { - string[] files = await GetFilesAsync(); + string file = message.fileName; - if (files == null || files.Length == 0) + if (await HasAlreadyBeenProcessedAsync(file)) { - statusService.StatusPublish(new StatusUpdateMessage($"No files found in '{fileLocations.offlocInput}'")); - stagingService.StagingPublish(new OfflocParserFinishedMessage("No Path", true)); + statusService.StatusPublish(new StatusUpdateMessage($"File {file} has already been processed")); + stagingService.StagingPublish(new OfflocParserFinishedMessage("File already processed", true)); } else { - await cleaningService.CleanFiles(files); + var request = new OfflocFileProcessingStarted(message.fileName, message.FileId, message.ArchiveFileName); + await dbService.SendDbRequestAndWaitForResponse(request); + await cleaningService.CleanFile(file); } } - - //Returns files that fit validation checks (ie. not already been processed). - private async Task GetFilesAsync() - { - string[] files = Directory.GetFiles(fileLocations.offlocInput).Where(f => !f.Contains("LineFeeds")).ToArray(); - - for (int i = 0; i < files.Length; i++) - { - files[i] = files[i][(fileLocations.offlocInput.Length + 1)..]; - } - - var res = await GetAlreadyProcessedFilesAsync(); - - return files.Where(f => !res.Contains(f)).ToArray(); - } - - private async Task GetAlreadyProcessedFilesAsync() + private async Task HasAlreadyBeenProcessedAsync(string file) { - var res = await dbService.DbTransientSubscribe(new GetOfflocFilesMessage()); - return res.offlocFiles; + var res = await dbService.SendDbRequestAndWaitForResponse(new GetOfflocFilesMessage()); + return res.offlocFiles.Contains(file); } } diff --git a/src/Offloc.Cleaner/Program.cs b/src/Offloc.Cleaner/Program.cs index 6749f2d..51e6364 100644 --- a/src/Offloc.Cleaner/Program.cs +++ b/src/Offloc.Cleaner/Program.cs @@ -28,16 +28,8 @@ builder.Configuration.GetValue("RedundantOfflocFields")! )); - bool parallel = builder.Configuration.GetValue("Parallel"); - if (parallel) - { - builder.Services.AddSingleton(); - } - else - { - builder.Services.AddSingleton(); - } - + builder.Services.AddSingleton(); + builder.Services.ConfigureServices(builder.Configuration); builder.Services.AddSingleton(); builder.Services.AddSingleton(); diff --git a/src/Offloc.Cleaner/Services/CleaningStrategyBase.cs b/src/Offloc.Cleaner/Services/CleaningStrategyBase.cs index b51a637..64a32c8 100644 --- a/src/Offloc.Cleaner/Services/CleaningStrategyBase.cs +++ b/src/Offloc.Cleaner/Services/CleaningStrategyBase.cs @@ -23,7 +23,7 @@ public CleaningStrategyBase(IStagingMessagingService stagingService, IStatusMess this.fileLocations = fileLocations; } - public async Task CleanFile(string file) + public virtual async Task ProcessFile(string file) { await Task.CompletedTask; diff --git a/src/Offloc.Cleaner/Services/ICleaningStrategy.cs b/src/Offloc.Cleaner/Services/ICleaningStrategy.cs index 0072b6a..67de296 100644 --- a/src/Offloc.Cleaner/Services/ICleaningStrategy.cs +++ b/src/Offloc.Cleaner/Services/ICleaningStrategy.cs @@ -3,5 +3,5 @@ namespace Offloc.Cleaner.Services; public interface ICleaningStrategy { - Task CleanFiles(string[] files); + Task CleanFile(string file); } diff --git a/src/Offloc.Cleaner/Services/ParallelCleaningStrategy.cs b/src/Offloc.Cleaner/Services/ParallelCleaningStrategy.cs deleted file mode 100644 index c5c4cfd..0000000 --- a/src/Offloc.Cleaner/Services/ParallelCleaningStrategy.cs +++ /dev/null @@ -1,35 +0,0 @@ - -using FileStorage; -using Messaging.Interfaces; -using Messaging.Messages.StagingMessages; -using Messaging.Messages.StatusMessages; - -namespace Offloc.Cleaner.Services; - -public class ParallelCleaningStrategy : CleaningStrategyBase, ICleaningStrategy -{ - public ParallelCleaningStrategy(IStagingMessagingService stagingService, - IStatusMessagingService statusService, IFileLocations fileLocations, - int[] redundantFields) - : base(stagingService, statusService, fileLocations, redundantFields) { } - - public async Task CleanFiles(string[] files) - { - await Task.CompletedTask; - - List tasks = new List(); - - Parallel.For(0, files.Length, (i, f) => { - statusService.StatusPublish(new StatusUpdateMessage($"Processing file: {files[i]}")); - Task t = CleanFile($"{fileLocations.offlocInput}/{files[i]}"); - tasks.Add(t); - }); - - Task.WaitAll(tasks.ToArray()); //Waits for all cleaning to finish so parser can run in parallel easily. - - if (files.Length > 0) - { - stagingService.StagingPublish(new OfflocCleanerFinishedMessage(files, redundantFieldIndexes)); - } - } -} diff --git a/src/Offloc.Cleaner/Services/SequentialCleaningStrategy.cs b/src/Offloc.Cleaner/Services/SequentialCleaningStrategy.cs index b3fecb6..c7fd251 100644 --- a/src/Offloc.Cleaner/Services/SequentialCleaningStrategy.cs +++ b/src/Offloc.Cleaner/Services/SequentialCleaningStrategy.cs @@ -14,33 +14,11 @@ public class SequentialCleaningStrategy : CleaningStrategyBase, ICleaningStrateg : base(stagingService, statusService, fileLocations, redundantWrapper.redundantFieldIndexes){ } - public async Task CleanFiles(string[] files) + public async Task CleanFile(string file) { - files = OrderFilesByDateAsc(files); - - for (int i = 0; i < files.Length; i++) - { - statusService.StatusPublish(new StatusUpdateMessage($"Cleaning file: {files[i]}")); - await CleanFile($"{fileLocations.offlocInput}/{files[i]}"); - stagingService.StagingPublish(new OfflocCleanerFinishedMessage(new string[] { files[i] }, redundantFieldIndexes)); - } + statusService.StatusPublish(new StatusUpdateMessage($"Cleaning file: {file}")); + await base.ProcessFile(Path.Combine(fileLocations.offlocInput, file)); + stagingService.StagingPublish(new OfflocCleanerFinishedMessage([file], redundantFieldIndexes)); } - //Ordering of files is only needed if sequential processing is toggled. - private string[] OrderFilesByDateAsc(string[] files) - { - Dictionary dictionary = new Dictionary(); - - foreach (var file in files) - { - string s = file.Substring(17, 8); - s = $"{s[..2]}/{s[2..4]}/{s[4..]}"; - DateTime datetime; - DateTime.TryParseExact(s, new[] { "dd/MM/yyyy" }, CultureInfo.InvariantCulture, DateTimeStyles.None, out datetime); - dictionary.Add(file, datetime); - } - dictionary = dictionary.OrderBy(d => d.Value).ToDictionary(); - - return dictionary.Keys.ToArray(); - } -} +} \ No newline at end of file diff --git a/src/Offloc.Cleaner/appsettings.json b/src/Offloc.Cleaner/appsettings.json index f1c8246..9686cab 100644 --- a/src/Offloc.Cleaner/appsettings.json +++ b/src/Offloc.Cleaner/appsettings.json @@ -1,7 +1,6 @@ { "Serilog:WriteTo:1:Args:path": "%USERPROFILE%/DMS/logs/Offloc.Cleaner-.txt", "EnvFilePath": "../", - "Parallel": false, //Indexing here assumes 0-indexing. Corresponding Id in Offloc definition is 1 greater. "fieldsToIgnore": [ 11, diff --git a/src/Offloc.Parser/Offloc.Parser.csproj b/src/Offloc.Parser/Offloc.Parser.csproj index 801c8dc..8f804e8 100644 --- a/src/Offloc.Parser/Offloc.Parser.csproj +++ b/src/Offloc.Parser/Offloc.Parser.csproj @@ -11,12 +11,10 @@ - - diff --git a/src/Offloc.Parser/Program.cs b/src/Offloc.Parser/Program.cs index 5212da6..4b8aee5 100644 --- a/src/Offloc.Parser/Program.cs +++ b/src/Offloc.Parser/Program.cs @@ -26,16 +26,8 @@ builder.Configuration.AddJsonFile("appsettings.json").AddEnvironmentVariables(); builder.Configuration.ConfigureByEnvironment(); - bool parallel = builder.Configuration.GetValue("Parallel"); - if (parallel) - { - builder.Services.AddSingleton(); - } - else - { - builder.Services.AddSingleton(); - } - + builder.Services.AddSingleton(); + //Extremely bad practice but just builder.Services.AddSingleton(builder.Configuration); diff --git a/src/Offloc.Parser/Services/ParsingStrategy/ParallelParsingStrategy.cs b/src/Offloc.Parser/Services/ParsingStrategy/ParallelParsingStrategy.cs deleted file mode 100644 index c86e2dc..0000000 --- a/src/Offloc.Parser/Services/ParsingStrategy/ParallelParsingStrategy.cs +++ /dev/null @@ -1,25 +0,0 @@ -using FileStorage; -using Messaging.Interfaces; -using Messaging.Messages.StagingMessages; -using Offloc.Parser.Services.TrimmerContext; - -namespace Offloc.Parser.Services; - -public class ParallelParsingStrategy : ParsingStrategyBase, IParsingStrategy -{ - public ParallelParsingStrategy(IStagingMessagingService stagingService, - IStatusMessagingService statusService, IFileLocations fileLocations, - FieldTrimmerContext trimmerContext) - : base(stagingService, statusService, fileLocations, trimmerContext){ } - - public async Task ParseFiles(string[] files) - { - await Task.CompletedTask; - - Parallel.For(0, files.Length, async (i, f) => - { - await ParseFile(fileLocations.offlocInput + '/' + files[i]); - stagingService.StagingPublish(new OfflocParserFinishedMessage(files[i].Split('/').Last(), false)); - }); - } -} \ No newline at end of file diff --git a/src/Offloc.Parser/appsettings.json b/src/Offloc.Parser/appsettings.json index 26a9ee5..64a3d25 100644 --- a/src/Offloc.Parser/appsettings.json +++ b/src/Offloc.Parser/appsettings.json @@ -1,13 +1,12 @@ { "Serilog:WriteTo:1:Args:path": "%USERPROFILE%/DMS/logs/Offloc.Parser-.txt", "EnvFilePath": "../", - "Parallel": false, - //"Logging": { - // "LogLevel": { - // "Default": "Warning", - // "Microsoft.Hosting.Lifetime": "Warning" - // } - //}, + "Logging": { + "LogLevel": { + "Default": "Warning", + "Microsoft.Hosting.Lifetime": "Warning" + } + }, //Indexing here assumes 0-indexing. Corresponding Id in Offloc definition is 1 greater. "fieldsToIgnore": [ 11, diff --git a/src/OfflocRunningPictureDb/Stored Procedures/MergeAll.sql b/src/OfflocRunningPictureDb/Stored Procedures/MergeAll.sql index 0321182..b9dd38e 100644 --- a/src/OfflocRunningPictureDb/Stored Procedures/MergeAll.sql +++ b/src/OfflocRunningPictureDb/Stored Procedures/MergeAll.sql @@ -1,4 +1,5 @@ CREATE PROCEDURE [OfflocRunningPicture].[MergeAll] + @fileName NVARCHAR(255) NULL AS BEGIN SET NOCOUNT ON; @@ -30,6 +31,13 @@ BEGIN --Non Temporal Table EXEC [OfflocRunningPicture].[MergeStandardisationReference] + IF(@fileName IS NOT NULL) + BEGIN + UPDATE [OfflocRunningPicture].[ProcessedFiles] + SET [Status] = 'Merged' + WHERE [FileName] = @fileName; + END; + COMMIT TRANSACTION END TRY BEGIN CATCH diff --git a/src/OfflocRunningPictureDb/Tables/ProcessedFiles.sql b/src/OfflocRunningPictureDb/Tables/ProcessedFiles.sql index 1a078ee..d4da914 100644 --- a/src/OfflocRunningPictureDb/Tables/ProcessedFiles.sql +++ b/src/OfflocRunningPictureDb/Tables/ProcessedFiles.sql @@ -2,6 +2,8 @@ CREATE TABLE [OfflocRunningPicture].[ProcessedFiles] ( [FileName] VARCHAR (50) NOT NULL, [FileId] INT NOT NULL, + [ArchiveFileName] VARCHAR (50) NULL, + [Status] VARCHAR(10) NULL, [ValidFrom] DATETIME2 (7) GENERATED ALWAYS AS ROW START DEFAULT (sysutcdatetime()) NOT NULL, [ValidTo] DATETIME2 (7) GENERATED ALWAYS AS ROW END DEFAULT (CONVERT([datetime2],'9999-12-31 23:59:59.9999999')) NOT NULL, CONSTRAINT [PK_ProcessedFiles] PRIMARY KEY ([FileId] ASC), diff --git a/src/OfflocStagingDb/Stored Procedures/Import.sql b/src/OfflocStagingDb/Stored Procedures/Import.sql index 154b8cf..e73abb1 100644 --- a/src/OfflocStagingDb/Stored Procedures/Import.sql +++ b/src/OfflocStagingDb/Stored Procedures/Import.sql @@ -72,13 +72,10 @@ BEGIN Declare @retMessage varchar(500); EXEC [OfflocStaging].[StandardiseData] @retMessage; - INSERT INTO [OfflocRunningPictureDb].[OfflocRunningPicture].[ProcessedFiles] - ( - FileName, - FileId - ) - VALUES - (@processedFile, CAST(@stringDate AS INT)); + UPDATE [OfflocRunningPictureDb].[OfflocRunningPicture].[ProcessedFiles] + SET [Status] = 'Imported' + WHERE FileName = @processedFile; + COMMIT TRANSACTION; END TRY BEGIN CATCH diff --git a/src/Orchestrator/GlobalUsings.cs b/src/Orchestrator/GlobalUsings.cs deleted file mode 100644 index ffef714..0000000 --- a/src/Orchestrator/GlobalUsings.cs +++ /dev/null @@ -1,8 +0,0 @@ -global using EnvironmentSetup; -global using Orchestrator; -global using Messaging.Interfaces; -global using Messaging.Services; -global using Microsoft.Extensions.Configuration; -global using Microsoft.Extensions.DependencyInjection; -global using Microsoft.Extensions.Hosting; -global using Serilog; \ No newline at end of file diff --git a/src/Orchestrator/Orchestrator.csproj b/src/Orchestrator/Orchestrator.csproj deleted file mode 100644 index ceecc87..0000000 --- a/src/Orchestrator/Orchestrator.csproj +++ /dev/null @@ -1,26 +0,0 @@ - - - - Exe - enable - enable - - - - - - - - - - - - - - - - Always - - - - diff --git a/src/Orchestrator/OrchestratorBackgroundService.cs b/src/Orchestrator/OrchestratorBackgroundService.cs deleted file mode 100644 index e3e9949..0000000 --- a/src/Orchestrator/OrchestratorBackgroundService.cs +++ /dev/null @@ -1,180 +0,0 @@ -using FileStorage; -using Messaging.Messages.DbMessages.Receiving; -using Messaging.Messages.DbMessages.Sending; -using Messaging.Messages.MatchingMessages.Clustering; -using Messaging.Queues; -using Microsoft.Extensions.Logging; -using System.Globalization; -using System.Management.Automation; -using System.Management.Automation.Runspaces; - -namespace Orchestrator; - -/// -/// This is an 'ad-hoc' service intended to aid the process of bulk loading files into DMS. -/// -public class OrchestratorBackgroundService( - ILogger logger, - IConfiguration configuration, - IMatchingMessagingService matchingMessagingService, - IDbMessagingService dbMessagingService, - IFileLocations fileLocations) : BackgroundService -{ - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - try - { - var files = configuration.GetValue("FilesLocation") ?? throw new Exception("Missing configuration section for FilesLocation"); - var scriptLocation = configuration.GetValue("ScriptLocation") ?? throw new Exception("Missing configuration section for ScriptLocation"); - - var offlocFilePattern = "*.dat"; - var deliusFilePattern = "*.txt"; - - if (TryReadScript(scriptLocation, out string? script) is false) - { - throw new Exception("Could not find script!"); - } - - // Create and open a Runspace - var runspace = RunspaceFactory.CreateRunspace(); - runspace.Open(); - - // Create a single PowerShell instance - var powershell = PowerShell.Create(); - powershell.Runspace = runspace; - powershell.AddScript(script, true); - - await Task.Run(() => - { - matchingMessagingService.MatchingSubscribe(async message => - { - logger.LogInformation($"{nameof(ClusteringPostProcessingFinishedMessage)} received."); - - var allOfflocFiles = GetAllFilesByPattern(files, offlocFilePattern); - var allDeliusFiles = GetAllFilesByPattern(files, deliusFilePattern); - - if (allOfflocFiles is { Length: 0 } || allDeliusFiles is { Length: 0 }) - { - logger.LogInformation($"Missing files in '{files}'. Skipping further execution..."); - return; - } - - var processedOfflocFiles = await GetAlreadyProcessedOfflocFilesAsync(); - var processedDeliusFiles = await GetAlreadyProcessedDeliusFilesAsync(); - - var unprocessedOfflocFiles = GetUnprocessedFiles(allOfflocFiles, processedOfflocFiles); - var unprocessedDeliusFiles = GetUnprocessedFiles(allDeliusFiles, processedDeliusFiles); - - logger.LogInformation("Finding unprocessed files..."); - - var offlocFileToProcess = unprocessedOfflocFiles.OrderBy(fileName => - { - var y = fileName.Key.Substring(17, 8); - return DateOnly.Parse($"{y[..2]}/{y[2..4]}/{y[4..]}"); - }).FirstOrDefault(); - - var deliusFileToProcess = unprocessedDeliusFiles.OrderBy(fileName => - { - return DateOnly.Parse($"{fileName.Key[27..29]}/{fileName.Key[25..27]}/{fileName.Key[21..25]}"); - }).FirstOrDefault(); - - logger.LogInformation("Find unprocessed files done."); - - if (offlocFileToProcess is { Key: null } || deliusFileToProcess is { Key: null }) - { - logger.LogInformation("No new file(s) to process. Skipping further execution..."); - return; - } - - logger.LogInformation($"Copying file '{offlocFileToProcess.Key}' to '{fileLocations.offlocInput}'..."); - File.Copy(offlocFileToProcess.Value, Path.Combine(fileLocations.offlocInput, offlocFileToProcess.Key), true); - logger.LogInformation("Copy file done."); - - logger.LogInformation($"Copying file '{deliusFileToProcess.Key}' to '{fileLocations.deliusInput}'..."); - File.Copy(deliusFileToProcess.Value, Path.Combine(fileLocations.deliusInput, deliusFileToProcess.Key), true); - logger.LogInformation("Copy file done."); - - await ExecuteScriptAsync(powershell, stoppingToken); - - }, TMatchingQueue.ClusteringPostProcessingFinished); - - }, stoppingToken); - - } - catch (Exception ex) - { - logger.LogError(ex, "Unexpected error occurred"); - } - - } - - private static bool TryReadScript(string location, out string script) - { - script = string.Empty; - - if(File.Exists(location) is false) - { - return false; - } - - script = File.ReadAllText(location); - - return true; - } - - private async Task ExecuteScriptAsync(PowerShell powershell, CancellationToken stoppingToken) - { - logger.LogInformation("Executing (powershell) script..."); - - try - { - await powershell.InvokeAsync(); - - foreach(var error in powershell.Streams.Error) - { - logger.LogError($"Powershell: {error}"); - } - - } - catch (Exception ex) - { - logger.LogError(ex, ex.Message); - } - - logger.LogInformation("Execution complete."); - } - - private static IEnumerable> GetUnprocessedFiles(string[] allFiles, string[] processedFiles) - { - return allFiles.ToDictionary(file => Path.GetFileName(file), filePath => filePath) - .Where(file => !processedFiles.Contains(file.Key)); - } - - private string[] GetAllFilesByPattern(string location, string pattern) - { - logger.LogInformation($"Retrieving '{pattern}' files from '{location}'..."); - var files = Directory.GetFiles(location, pattern, SearchOption.AllDirectories); - logger.LogInformation("Retrieving files done."); - - return files; - } - - private async Task GetAlreadyProcessedOfflocFilesAsync() - { - logger.LogInformation("Getting processed Offloc files..."); - var offloc = await dbMessagingService.DbTransientSubscribe(new GetOfflocFilesMessage()); - logger.LogInformation("Get processed files done."); - - return offloc.offlocFiles; - } - - private async Task GetAlreadyProcessedDeliusFilesAsync() - { - logger.LogInformation("Getting processed Delius files..."); - var delius = await dbMessagingService.DbTransientSubscribe(new GetDeliusFilesMessage()); - logger.LogInformation("Get processed files done."); - - return delius.fileNames; - } - -} diff --git a/src/Orchestrator/Program.cs b/src/Orchestrator/Program.cs deleted file mode 100644 index 44aeae8..0000000 --- a/src/Orchestrator/Program.cs +++ /dev/null @@ -1,36 +0,0 @@ -Log.Logger = new LoggerConfiguration() - .Enrich.FromLogContext() - .WriteTo.Console() - .WriteTo.File(@".\logs\fatal.txt", Serilog.Events.LogEventLevel.Fatal) - .CreateBootstrapLogger(); - -try -{ - var builder = Host.CreateApplicationBuilder(args); - - builder.Configuration.ConfigureByEnvironment(); - - builder.Services.ConfigureServices(builder.Configuration); - - builder.Services.AddSingleton(); - builder.Services.AddSingleton(); - builder.Services.AddSingleton(); - - builder.Services.AddHostedService(); - - var app = builder.Build(); - Log.Information("Starting application"); - app.Run(); - - return 0; -} -catch (Exception ex) -{ - Log.Fatal(ex, "Host terminated unexpectedly"); - return 1; -} -finally -{ - Log.Information("Application stopping"); - await Log.CloseAndFlushAsync(); -} \ No newline at end of file diff --git a/src/Orchestrator/Properties/launchSettings.json b/src/Orchestrator/Properties/launchSettings.json deleted file mode 100644 index b0398c3..0000000 --- a/src/Orchestrator/Properties/launchSettings.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "$schema": "http://json.schemastore.org/launchsettings.json", - "profiles": { - "Orchestrator": { - "commandName": "Project", - "dotnetRunMessages": true, - "environmentVariables": { - "DOTNET_ENVIRONMENT": "Development" - } - } - } -} diff --git a/src/Orchestrator/appsettings.Development.json b/src/Orchestrator/appsettings.Development.json deleted file mode 100644 index b2dcdb6..0000000 --- a/src/Orchestrator/appsettings.Development.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "Logging": { - "LogLevel": { - "Default": "Information", - "Microsoft.Hosting.Lifetime": "Information" - } - } -} diff --git a/src/Orchestrator/appsettings.json b/src/Orchestrator/appsettings.json deleted file mode 100644 index 416d702..0000000 --- a/src/Orchestrator/appsettings.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "Serilog:WriteTo:1:Args:path": "%USERPROFILE%/DMS/logs/Orchestrator-.txt", - "FilesLocation": "C:\\path\\to\\files\\", - "ScriptLocation": "C:\\path\\to\\script.ps1" -} diff --git a/src/Visualiser/Visualiser.csproj b/src/Visualiser/Visualiser.csproj index cefedde..623fd3c 100644 --- a/src/Visualiser/Visualiser.csproj +++ b/src/Visualiser/Visualiser.csproj @@ -1,7 +1,6 @@  - net10.0 enable enable Visualiser @@ -16,7 +15,7 @@ - + \ No newline at end of file diff --git a/src/Visualiser/wwwroot/js/network.js b/src/Visualiser/wwwroot/js/network.js index 1e806a9..d5b0e12 100644 --- a/src/Visualiser/wwwroot/js/network.js +++ b/src/Visualiser/wwwroot/js/network.js @@ -460,7 +460,7 @@ function reassignNode(node, toClusterId) { let edges = dataset.edges.get({ filter: (edge) => edge.from === node.id || edge.to === node.id }); edges.forEach(edge => dataset.edges.remove(edge.id)); - const fromClusterId = node.id; + const fromClusterId = node.group; // Destroy target cluster if (isCluster(toClusterId)) {