Feature/gh00839 use two web job instances to create messages to azure service bus based on content in busevent table

azure-pipeline.yml

@@ -12,7 +12,6 @@ resources:
       name: equinor/procosys-infra
       endpoint: 'equinor'
       ref: master
-
 variables:
   - template: src/variables/global-variables.yml@templates
   - template: src/resources/bus-sender/bus-sender-variables.yml@templates

src/Equinor.ProCoSys.BusSender.Core/DateTimeConverter.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core;
+public class DateTimeConverter : JsonConverter<DateTime?>
+{
+    private const string DateFormat = "yyyy-MM-ddTHH:mm:ss";
+
+    public override DateTime? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
+    {
+        if (reader.TokenType == JsonTokenType.Null)
+        {
+            return null;
+        }
+
+        if (reader.TokenType == JsonTokenType.String)
+        {
+            if (DateTime.TryParseExact(reader.GetString(), DateFormat, null, System.Globalization.DateTimeStyles.None, out DateTime dateTime))
+            {
+                return dateTime;
+            }
+        }
+
+        throw new JsonException();
+    }
+
+    public override void Write(Utf8JsonWriter writer, DateTime? value, JsonSerializerOptions options)
+    {
+        if (value.HasValue)
+        {
+            writer.WriteStringValue(value.Value.ToString(DateFormat));
+        }
+        else
+        {
+            writer.WriteNullValue();
+        }
+    }
+}

src/Equinor.ProCoSys.BusSender.Core/Equinor.ProCoSys.BusSenderWorker.Core.csproj

@@ -7,9 +7,17 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Azure.Data.AppConfiguration" Version="1.2.0" />
+    <PackageReference Include="Azure.Identity" Version="1.13.2" />
+    <PackageReference Include="Azure.Storage.Blobs" Version="12.18.0" />
     <PackageReference Include="Microsoft.ApplicationInsights" Version="2.21.0" />
+    <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.1.1" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="9.0.1" />
     <PackageReference Include="Oracle.EntityFrameworkCore" Version="7.21.12" />
+
+    <PackageReference Include="Polly" Version="8.5.2" />
     <PackageReference Include="System.Text.Json" Version="8.0.5" />
+
   </ItemGroup>
 
   <ItemGroup>

src/Equinor.ProCoSys.BusSender.Core/Interfaces/IBlobLeaseService.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Equinor.ProCoSys.BusSenderWorker.Core.Models;
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core.Interfaces;
+public interface IBlobLeaseService
+{
+    void ReleasePlantLease(PlantLease? plantLease, int maxRetryAttempts = 0);
+    Task<List<PlantLease>?> ClaimPlantLease();
+    CancellationToken CancellationToken { get; }
+}

src/Equinor.ProCoSys.BusSender.Core/Interfaces/IBusEventRepository.cs

@@ -10,4 +10,5 @@ public interface IBusEventRepository
     Task<List<BusEvent>> GetEarliestUnProcessedEventChunk();
     Task<long> GetUnProcessedCount();
     Task<DateTime> GetOldestEvent();
+    void SetPlants(List<string> plants);
 }

src/Equinor.ProCoSys.BusSender.Core/Interfaces/IPlantRepository.cs

@@ -0,0 +1,9 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core.Interfaces;
+
+public interface IPlantRepository
+{
+    List<string> GetAllPlants();
+}

src/Equinor.ProCoSys.BusSender.Core/Interfaces/IPlantService.cs

@@ -0,0 +1,10 @@
+using System.Collections.Generic;
+using Equinor.ProCoSys.BusSenderWorker.Core.Models;
+using Microsoft.Extensions.Configuration;
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core.Interfaces;
+public interface IPlantService
+{
+    List<string> GetPlantsHandledByInstance(List<PlantLease> plantLeases);
+    List<string>? GetAllPlants();
+}

src/Equinor.ProCoSys.BusSender.Core/Interfaces/IUnitOfWork.cs

@@ -1,8 +1,10 @@
-using System.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Equinor.ProCoSys.BusSenderWorker.Core.Interfaces;
 
 public interface IUnitOfWork
 {
     Task<int> SaveChangesAsync();
+    Task<int> SaveChangesAsync(CancellationToken cancellationToken);
 }

src/Equinor.ProCoSys.BusSender.Core/Models/Plant.cs

@@ -0,0 +1,10 @@
+#pragma warning disable CS8618
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core.Models;
+
+public class Plant
+{
+    public string ProjectSchema { get; set; }
+    public string IsVoided { get; set; }
+}
+

src/Equinor.ProCoSys.BusSender.Core/Models/PlantLease.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Text.Json.Serialization;
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core.Models;
+public class PlantLease
+{
+    [JsonConverter(typeof(DateTimeConverter))]
+    public DateTime? LeaseExpiry { get; set; }
+    public required string Plant { get; set; }
+    [JsonConverter(typeof(DateTimeConverter))]
+    public DateTime? LastProcessed { get; set; }
+
+    [JsonIgnore]
+    // Flag to tell if this plant is the one being handled by the current instance.
+    // This is transient and not stored in the blob.
+    public bool IsCurrent { get; set; } = false;
+}

src/Equinor.ProCoSys.BusSender.Core/Services/BlobLeaseService.cs

@@ -0,0 +1,240 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Text.Json;
+using System.Threading.Tasks;
+using Azure.Storage.Blobs.Models;
+using Azure.Storage.Blobs;
+using Azure;
+using Azure.Storage.Blobs.Specialized;
+using Equinor.ProCoSys.BusSenderWorker.Core.Interfaces;
+using Equinor.ProCoSys.BusSenderWorker.Core.Models;
+using Polly;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Configuration;
+using System.Threading;
+
+namespace Equinor.ProCoSys.BusSenderWorker.Core.Services;
+
+public class BlobLeaseService : IBlobLeaseService
+{
+    private readonly ILogger<BlobLeaseService> _logger;
+    private readonly IConfiguration _configuration;
+    private CancellationTokenSource? _cancellationTokenSource;
+
+    public CancellationToken CancellationToken => _cancellationTokenSource?.Token ?? CancellationToken.None;
+
+    public BlobLeaseService(ILogger<BlobLeaseService> logger, IConfiguration configuration)
+    {
+        _logger = logger;
+        _configuration = configuration;
+    }
+
+    public async Task<List<PlantLease>?> ClaimPlantLease()
+    {
+        if (!int.TryParse(_configuration["BlobLeaseExpiryTime"], out var blobLeaseExpiryTime))
+        {
+            _logger.LogError("Invalid BlobLeaseExpiryTime configuration value.");
+            return null;
+        }
+
+        if (_cancellationTokenSource == null || _cancellationTokenSource.Token == CancellationToken.None)
+        {
+            // Multiplying by a factor lower than 1 to ensure that message processing for this instance is cancelled shortly before the lease expires.
+            _cancellationTokenSource = new CancellationTokenSource((int)(blobLeaseExpiryTime * 1000 * 0.95));
+        }
+
+        var leaseId = Guid.NewGuid().ToString();
+        var plantLeases = await GetPlantLeases(leaseId);
+        if (plantLeases == null)
+        {
+            // Nothing to do for now.
+            return null;
+        }
+
+        var plantLease = GetOldestUnprocessedPlantLeaseInfo(plantLeases);
+        if (plantLease == null)
+        {
+            // Nothing to do for now.
+            return null;
+        }
+
+        plantLeases.Where(x => x.Plant.Equals(plantLease.Plant)).ToList().ForEach(x =>
+        {
+            x.IsCurrent = true;
+            x.LeaseExpiry = DateTime.UtcNow.AddSeconds(blobLeaseExpiryTime);
+        });
+
+        UpdatePlantLeases(plantLeases, leaseId);
+        return plantLeases;
+    }
+
+    public async void ReleasePlantLease(PlantLease? plantLease, int maxRetryAttempts = 0)
+    {
+        if (plantLease == null)
+        {
+            _logger.LogWarning("Plant lease is null. Cannot release lease.");
+            return;
+        }
+
+        var leaseId = Guid.NewGuid().ToString();
+        var plantLeases = await GetPlantLeases(leaseId,maxRetryAttempts);
+        if (plantLeases == null)
+        {
+            _logger.LogWarning("Plant leases is null. Cannot release lease.");
+            return;
+        }
+
+        plantLeases.Where(x => x.Plant.Equals(plantLease?.Plant)).ToList().ForEach(x =>
+        {
+            x.LeaseExpiry = null;
+            x.LastProcessed = DateTime.UtcNow;
+        });
+        UpdatePlantLeases(plantLeases, leaseId);
+    }
+
+    public virtual BlobLeaseClient GetBlobLeaseClient(BlobClient blobClient, string leaseId) => blobClient.GetBlobLeaseClient(leaseId);
+
+    protected async Task<bool> TryAcquireBlobLeaseAsync(BlobClient blobClient, string leaseId, TimeSpan leaseDuration, int maxRetryAttempts = 0, TimeSpan? delayBetweenAttempts = null)
+    {
+        if (!int.TryParse(_configuration["BlobReleaseLeaseDelay"], out var blobReleaseLeaseDelay))
+        {
+            _logger.LogError("Invalid BlobReleaseLeaseDelay configuration value.");
+        }
+        delayBetweenAttempts ??= TimeSpan.FromSeconds(blobReleaseLeaseDelay);
+
+        var retryPolicy = Policy
+            .Handle<RequestFailedException>(ex => ex.ErrorCode == BlobErrorCode.LeaseAlreadyPresent)
+            .WaitAndRetryAsync(maxRetryAttempts, retryAttempt =>
+            {
+                _logger.LogInformation($"Attempt {retryAttempt} to acquire lease for blob: {blobClient.Name}");
+                return delayBetweenAttempts.Value;
+            });
+
+        try
+        {
+            await retryPolicy.ExecuteAsync(async () =>
+            {
+                var leaseClient = GetBlobLeaseClient(blobClient, leaseId);
+                await leaseClient.AcquireAsync(leaseDuration, cancellationToken: CancellationToken.None);
+            });
+            return true;
+        }
+        catch (RequestFailedException rfe)
+        {
+            if (rfe.ErrorCode == BlobErrorCode.LeaseAlreadyPresent)
+            {
+                return false;
+            }
+
+            _logger.LogError(rfe, $"Failed to acquire lease for blob: {blobClient.Name} ErrorCode: {rfe.ErrorCode} Message: {rfe.Message}");
+            throw;
+        }
+    }
+
+    public virtual async Task<string> GetBlobContentAsync(BlobClient blobClient)
+    {
+        try
+        {
+            var response = await blobClient.DownloadStreamingAsync(cancellationToken: CancellationToken.None);
+            using var streamReader = new StreamReader(response.Value.Content);
+            return await streamReader.ReadToEndAsync();
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Failed to download blob content: {BlobName}", blobClient.Name);
+            throw;
+        }
+    }
+
+    private async Task<List<PlantLease>?> GetPlantLeases(string leaseId, int maxRetryAttempts = 0)
+    {
+        var blobClient = GetBlobClient();
+        var newLeaseAcquired = await TryAcquireBlobLeaseAsync(blobClient, leaseId, TimeSpan.FromSeconds(15), maxRetryAttempts);
+        if (newLeaseAcquired)
+        {
+            var plantLease = await GetPlantLeases(blobClient);
+            if (plantLease == null || !plantLease.Any())
+            {
+                _logger.LogWarning("Could not read blob containing plant lease.");
+                return null;
+            }
+
+            return plantLease;
+        }
+
+        return null;
+    }
+
+    private async Task<List<PlantLease>?> GetPlantLeases(BlobClient blobClient)
+    {
+        var options = new JsonSerializerOptions
+        {
+            PropertyNameCaseInsensitive = true,
+            Converters = { new DateTimeConverter() }
+        };
+
+        var json = await GetBlobContentAsync(blobClient);
+        return JsonSerializer.Deserialize<List<PlantLease>>(json, options);
+    }
+
+    public virtual async void UpdatePlantLeases(List<PlantLease> plantLeases, string leaseId)
+    {
+        var blobClient = GetBlobClient();
+        var options = new JsonSerializerOptions
+        {
+            WriteIndented = true,
+            PropertyNameCaseInsensitive = true,
+            Converters = { new DateTimeConverter() }
+        };
+
+        var json = JsonSerializer.Serialize(plantLeases, options);
+        using var memoryStream = new MemoryStream(Encoding.UTF8.GetBytes(json));
+        var uploadOptions = new BlobUploadOptions
+        {
+            Conditions = new BlobRequestConditions
+            {
+                LeaseId = leaseId
+            }
+        };
+        await blobClient.UploadAsync(memoryStream, uploadOptions, CancellationToken.None);
+        await ReleaseBlobLeaseAsync(blobClient, leaseId);
+    }
+
+    public virtual BlobClient GetBlobClient()
+    {
+        var connectionString = _configuration["BlobStorage:ConnectionString"];
+        var containerName = _configuration["BlobStorage:BusSenderContainerName"];
+        var blobLeaseFileName = _configuration["BlobLeaseFileName"];
+        var blobContainerClient = new BlobContainerClient(connectionString, containerName);
+        return blobContainerClient.GetBlobClient(blobLeaseFileName);
+    }
+
+    private PlantLease? GetOldestUnprocessedPlantLeaseInfo(List<PlantLease> plantLeases)
+    {
+        var unprocessedLeaseInfos = plantLeases
+            .Where(p => p.LeaseExpiry == null || p.LeaseExpiry < DateTime.UtcNow)
+            .OrderBy(p => p.LastProcessed)
+            .ToList();
+
+        return unprocessedLeaseInfos.FirstOrDefault(); // Return first if none is taken.
+    }
+    public virtual async Task ReleaseBlobLeaseAsync(BlobClient blobClient, string leaseId)
+    {
+        try
+        {
+            var leaseClient = GetBlobLeaseClient(blobClient, leaseId);
+            await leaseClient.ReleaseAsync(cancellationToken: CancellationToken);
+        }
+        catch (RequestFailedException ex)
+        {
+            _logger.LogError(ex, "Failed to release lease for blob: {BlobName}", blobClient.Name);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Unexpected error occurred while releasing lease for blob: {BlobName}", blobClient.Name);
+        }
+    }
+}