Refactor LinkIndexProvider into own project

Author: reakaleek

docs-builder.sln

@@ -99,6 +99,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "assembler-config-validate",
 		actions\assembler-config-validate\action.yml = actions\assembler-config-validate\action.yml
 	EndProjectSection
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elastic.Documentation.LinkIndex", "src\Elastic.Documentation.LinkIndex\Elastic.Documentation.LinkIndex.csproj", "{FD1AC230-798B-4AB9-8CE6-A06264885DBC}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -160,6 +162,10 @@ Global
 		{CD94F9E4-7FCD-4152-81F1-4288C6B75367}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{CD94F9E4-7FCD-4152-81F1-4288C6B75367}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{CD94F9E4-7FCD-4152-81F1-4288C6B75367}.Release|Any CPU.Build.0 = Release|Any CPU
+		{FD1AC230-798B-4AB9-8CE6-A06264885DBC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{FD1AC230-798B-4AB9-8CE6-A06264885DBC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{FD1AC230-798B-4AB9-8CE6-A06264885DBC}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{FD1AC230-798B-4AB9-8CE6-A06264885DBC}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{4D198E25-C211-41DC-9E84-B15E89BD7048} = {BE6011CC-1200-4957-B01F-FCCA10C5CF5A}
@@ -184,5 +190,6 @@ Global
 		{7D36DDDA-9E0B-4D2C-8033-5D62FF8B6166} = {059E787F-85C1-43BE-9DD6-CE319E106383}
 		{FB1C1954-D8E2-4745-BA62-04DD82FB4792} = {245023D2-D3CA-47B9-831D-DAB91A2FFDC7}
 		{E20FEEF9-1D1A-4CDA-A546-7FDC573BE399} = {245023D2-D3CA-47B9-831D-DAB91A2FFDC7}
+		{FD1AC230-798B-4AB9-8CE6-A06264885DBC} = {BE6011CC-1200-4957-B01F-FCCA10C5CF5A}
 	EndGlobalSection
 EndGlobal

src/Elastic.Documentation.LinkIndex/Elastic.Documentation.LinkIndex.csproj

@@ -0,0 +1,17 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net9.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <PackageReference Include="AWSSDK.S3" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\Elastic.Documentation\Elastic.Documentation.csproj" />
+    </ItemGroup>
+
+</Project>

src/Elastic.Documentation.LinkIndex/LinkIndexProvider.cs

@@ -0,0 +1,62 @@
+using System.Net;
+using System.Runtime.CompilerServices;
+using Amazon.S3;
+using Amazon.S3.Model;
+using Elastic.Documentation.Links;
+using Microsoft.Extensions.Logging;
+
+namespace Elastic.Documentation.LinkIndex;
+
+public interface ILinkIndexProvider
+{
+	Task<LinkReferenceRegistry> GetLinkIndex(Cancel cancellationToken = default);
+	Task SaveLinkIndex(LinkReferenceRegistry registry, Cancel cancellationToken = default);
+	Task<LinkReference> GetLinkReference(string key, Cancel cancellationToken = default);
+}
+
+public class AwsS3LinkIndexProvider(IAmazonS3 s3Client, string bucketName, string registryKey) : ILinkIndexProvider
+{
+	public async Task<LinkReferenceRegistry> GetLinkIndex(Cancel cancellationToken = default)
+	{
+		var getObjectRequest = new GetObjectRequest
+		{
+			BucketName = bucketName,
+			Key = registryKey
+		};
+		var getObjectResponse = await s3Client.GetObjectAsync(getObjectRequest, cancellationToken);
+		await using var stream = getObjectResponse.ResponseStream;
+		var linkIndex = LinkReferenceRegistry.Deserialize(stream);
+		return linkIndex with { ETag = getObjectResponse.ETag };
+	}
+
+	public async Task SaveLinkIndex(LinkReferenceRegistry registry, Cancel cancellationToken = default)
+	{
+		if (registry.ETag == null)
+			// The ETag should not be null if the LinkReferenceRegistry was retrieved from GetLinkIndex()
+			throw new InvalidOperationException($"{nameof(LinkReferenceRegistry)}.{nameof(registry.ETag)} cannot be null");
+		var json = LinkReferenceRegistry.Serialize(registry);
+		var putObjectRequest = new PutObjectRequest
+		{
+			BucketName = bucketName,
+			Key = registryKey,
+			ContentBody = json,
+			ContentType = "application/json",
+			IfMatch = registry.ETag // Only update if the ETag matches. Meaning the object has not been changed in the meantime.
+		};
+		var putResponse = await s3Client.PutObjectAsync(putObjectRequest, cancellationToken);
+		if (putResponse.HttpStatusCode != HttpStatusCode.OK)
+			throw new Exception($"Unable to save {nameof(LinkReferenceRegistry)} to s3://{bucketName}/{registryKey}");
+	}
+
+	public async Task<LinkReference> GetLinkReference(string key, Cancel cancellationToken)
+	{
+		var getObjectRequest = new GetObjectRequest
+		{
+			BucketName = bucketName,
+			Key = key
+		};
+		var getObjectResponse = await s3Client.GetObjectAsync(getObjectRequest, cancellationToken);
+		await using var stream = getObjectResponse.ResponseStream;
+		return LinkReference.Deserialize(stream);
+	}
+}

src/Elastic.Documentation/Links/LinkReferenceRegistry.cs

@@ -14,6 +14,40 @@ public record LinkReferenceRegistry
 	[JsonPropertyName("repositories")]
 	public required Dictionary<string, Dictionary<string, LinkRegistryEntry>> Repositories { get; init; }
 
+	[JsonIgnore]
+	public required string? ETag { get; init; }
+
+	public LinkReferenceRegistry WithLinkRegistryEntry(LinkRegistryEntry entry)
+	{
+		var copiedRepositories = new Dictionary<string, Dictionary<string, LinkRegistryEntry>>(Repositories);
+		var repository = entry.Repository;
+		var branch = entry.Branch;
+		// repository already exists in links.json
+		if (copiedRepositories.TryGetValue(repository, out var existingRepositoryEntry))
+		{
+			// The branch already exists in the repository entry
+			if (existingRepositoryEntry.TryGetValue(branch, out var existingBranchEntry))
+			{
+				if (entry.UpdatedAt > existingBranchEntry.UpdatedAt)
+					existingRepositoryEntry[branch] = entry;
+			}
+			// branch does not exist in the repository entry
+			else
+			{
+				existingRepositoryEntry[branch] = entry;
+			}
+		}
+		// onboarding new repository
+		else
+		{
+			copiedRepositories.Add(repository, new Dictionary<string, LinkRegistryEntry>
+			{
+				{ branch, entry }
+			});
+		}
+		return this with { Repositories = copiedRepositories };
+	}
+
 	public static LinkReferenceRegistry Deserialize(Stream json) =>
 		JsonSerializer.Deserialize(json, SourceGenerationContext.Default.LinkReferenceRegistry)!;
 
@@ -46,4 +80,3 @@ public record LinkRegistryEntry
 	[JsonPropertyName("updated_at")]
 	public DateTime UpdatedAt { get; init; } = DateTime.MinValue;
 }
-

src/infra/docs-lambda-index-publisher/LinkIndexProvider.cs

@@ -11,6 +11,7 @@
 using Amazon.S3;
 using Amazon.S3.Util;
 using Elastic.Documentation.Lambda.LinkIndexUploader;
+using Elastic.Documentation.LinkIndex;
 using Elastic.Documentation.Links;
 
 const string bucketName = "elastic-docs-link-index";
@@ -26,19 +27,22 @@
 static async Task<SQSBatchResponse> Handler(SQSEvent ev, ILambdaContext context)
 {
 	var s3Client = new AmazonS3Client();
-	var linkIndexProvider = new LinkIndexProvider(s3Client, context.Logger, bucketName, indexFile);
+	ILinkIndexProvider linkIndexProvider = new AwsS3LinkIndexProvider(s3Client, bucketName, indexFile);
 	var batchItemFailures = new List<SQSBatchResponse.BatchItemFailure>();
+
+	var linkRegistry = await linkIndexProvider.GetLinkIndex();
+
 	foreach (var message in ev.Records)
 	{
 		context.Logger.LogInformation("Processing message {MessageId}", message.MessageId);
 		context.Logger.LogInformation("Message body: {MessageBody}", message.Body);
 		try
 		{
-			var s3RecordLinkReferenceTuples = await GetS3RecordLinkReferenceTuples(s3Client, message, context);
+			var s3RecordLinkReferenceTuples = await GetS3RecordLinkReferenceTuples(linkIndexProvider, message);
 			foreach (var (s3Record, linkReference) in s3RecordLinkReferenceTuples)
 			{
 				var newEntry = ConvertToLinkIndexEntry(s3Record, linkReference);
-				await linkIndexProvider.UpdateLinkIndexEntry(newEntry);
+				linkRegistry = linkRegistry.WithLinkRegistryEntry(newEntry);
 			}
 		}
 		catch (Exception e)
@@ -53,7 +57,7 @@ static async Task<SQSBatchResponse> Handler(SQSEvent ev, ILambdaContext context)
 	}
 	try
 	{
-		await linkIndexProvider.Save();
+		await linkIndexProvider.SaveLinkIndex(linkRegistry);
 		var response = new SQSBatchResponse(batchItemFailures);
 		if (batchItemFailures.Count > 0)
 			context.Logger.LogInformation("Failed to process {batchItemFailuresCount} of {allMessagesCount} messages. Returning them to the queue.", batchItemFailures.Count, ev.Records.Count);
@@ -94,15 +98,14 @@ static LinkRegistryEntry ConvertToLinkIndexEntry(S3EventNotification.S3EventNoti
 	};
 }
 
-static async Task<IReadOnlyCollection<(S3EventNotification.S3EventNotificationRecord, LinkReference)>> GetS3RecordLinkReferenceTuples(IAmazonS3 s3Client,
-	SQSEvent.SQSMessage message, ILambdaContext context)
+static async Task<IReadOnlyCollection<(S3EventNotification.S3EventNotificationRecord, LinkReference)>> GetS3RecordLinkReferenceTuples(ILinkIndexProvider linkIndexRepository,
+	SQSEvent.SQSMessage message)
 {
 	var s3Event = S3EventNotification.ParseJson(message.Body);
 	var recordLinkReferenceTuples = new ConcurrentBag<(S3EventNotification.S3EventNotificationRecord, LinkReference)>();
-	var linkReferenceProvider = new LinkReferenceProvider(s3Client, context.Logger, bucketName);
 	await Parallel.ForEachAsync(s3Event.Records, async (record, ctx) =>
 	{
-		var linkReference = await linkReferenceProvider.GetLinkReference(record.S3.Object.Key, ctx);
+		var linkReference = await linkIndexRepository.GetLinkReference(record.S3.Object.Key, ctx);
 		recordLinkReferenceTuples.Add((record, linkReference));
 	});
 	return recordLinkReferenceTuples;

src/infra/docs-lambda-index-publisher/LinkReferenceProvider.cs

@@ -29,6 +29,7 @@
     <PackageReference Include="AWSSDK.SQS" />
   </ItemGroup>
   <ItemGroup>
+    <ProjectReference Include="..\..\Elastic.Documentation.LinkIndex\Elastic.Documentation.LinkIndex.csproj" />
     <ProjectReference Include="..\..\Elastic.Documentation\Elastic.Documentation.csproj"/>
   </ItemGroup>
 </Project>