Refactor

Author: reakaleek

src/infra/docs-lambda-index-publisher/LinkIndexUpdater.cs renamed to src/infra/docs-lambda-index-publisher/LinkIndexProvider.cs

@@ -9,10 +9,18 @@
 
 namespace Elastic.Documentation.Lambda.LinkIndexUploader;
 
-public class LinkIndexUpdater(IAmazonS3 s3Client, ILambdaLogger logger, string bucketName, string key)
+/// <summary>
+/// Gets the link index from S3 once.
+/// You can then update the link index with <see cref="UpdateLinkIndexEntry(LinkIndexEntry)"/> and save it with <see cref="Save()"/>.
+/// If the link index changed in the meantime, <see cref="Save()"/> will throw an exception,
+/// thus all the messages from the queue will be sent back to the queue.
+/// </summary>
+public class LinkIndexProvider(IAmazonS3 s3Client, ILambdaLogger logger, string bucketName, string key)
 {
 	private string? _etag;
-	public async Task<LinkIndex> GetLinkIndex()
+	private LinkIndex? _linkIndex;
+
+	private async Task<LinkIndex> GetLinkIndex()
 	{
 		var getObjectRequest = new GetObjectRequest
 		{
@@ -24,12 +32,14 @@ public async Task<LinkIndex> GetLinkIndex()
 		await using var stream = getObjectResponse.ResponseStream;
 		_etag = getObjectResponse.ETag;
 		logger.LogInformation("Successfully got link index from s3://{bucketName}/{key}", bucketName, key);
-		return LinkIndex.Deserialize(stream);
+		_linkIndex = LinkIndex.Deserialize(stream);
+		return _linkIndex;
 	}
 
-	public void UpdateLinkIndexEntry(LinkIndex linkIndex, LinkIndexEntry linkIndexEntry)
+	public async Task UpdateLinkIndexEntry(LinkIndexEntry linkIndexEntry)
 	{
-		if (linkIndex.Repositories.TryGetValue(linkIndexEntry.Repository, out var existingEntry))
+		_linkIndex ??= await GetLinkIndex();
+		if (_linkIndex.Repositories.TryGetValue(linkIndexEntry.Repository, out var existingEntry))
 		{
 			var newEntryIsNewer = DateTime.Compare(linkIndexEntry.UpdatedAt, existingEntry[linkIndexEntry.Branch].UpdatedAt) > 0;
 			if (newEntryIsNewer)
@@ -42,20 +52,19 @@ public void UpdateLinkIndexEntry(LinkIndex linkIndex, LinkIndexEntry linkIndexEn
 		}
 		else
 		{
-			linkIndex.Repositories.Add(linkIndexEntry.Repository, new Dictionary<string, LinkIndexEntry>
+			_linkIndex.Repositories.Add(linkIndexEntry.Repository, new Dictionary<string, LinkIndexEntry>
 			{
 				{ linkIndexEntry.Branch, linkIndexEntry }
 			});
 			logger.LogInformation("Added new entry for {repository}@{branch}", linkIndexEntry.Repository, linkIndexEntry.Branch);
 		}
 	}
 
-	public async Task SaveLinkIndex(LinkIndex linkIndex)
+	public async Task Save()
 	{
-		if (_etag == null)
-			throw new InvalidOperationException("You must call GetLinkIndex() before SaveLinkIndex()");
-
-		var json = LinkIndex.Serialize(linkIndex);
+		if (_etag == null || _linkIndex == null)
+			throw new InvalidOperationException("You must call UpdateLinkIndexEntry() before Save()");
+		var json = LinkIndex.Serialize(_linkIndex);
 		logger.LogInformation("Saving link index to s3://{bucketName}/{key}", bucketName, key);
 		var putObjectRequest = new PutObjectRequest
 		{

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

@@ -22,12 +22,12 @@
 
 return;
 
+// The SQS queue is configured to trigger when elastic/*/*/links.json files are created or updated.
 static async Task<SQSBatchResponse> Handler(SQSEvent ev, ILambdaContext context)
 {
 	var s3Client = new AmazonS3Client();
-	var linkIndexUpdater = new LinkIndexUpdater(s3Client, context.Logger, bucketName, indexFile);
+	var linkIndexProvider = new LinkIndexProvider(s3Client, context.Logger, bucketName, indexFile);
 	var batchItemFailures = new List<SQSBatchResponse.BatchItemFailure>();
-	var linkIndex = await linkIndexUpdater.GetLinkIndex();
 	foreach (var message in ev.Records)
 	{
 		try
@@ -36,12 +36,12 @@ static async Task<SQSBatchResponse> Handler(SQSEvent ev, ILambdaContext context)
 			foreach (var (s3Record, linkReference) in s3RecordLinkReferenceTuples)
 			{
 				var newEntry = ConvertToLinkIndexEntry(s3Record, linkReference);
-				linkIndexUpdater.UpdateLinkIndexEntry(linkIndex, newEntry);
+				await linkIndexProvider.UpdateLinkIndexEntry(newEntry);
 			}
 		}
 		catch (Exception e)
 		{
-			//Add failed message identifier to the batchItemFailures list
+			// Add failed message identifier to the batchItemFailures list
 			context.Logger.LogWarning(e, "Failed to process message {MessageId}", message.MessageId);
 			batchItemFailures.Add(new SQSBatchResponse.BatchItemFailure
 			{
@@ -51,17 +51,18 @@ static async Task<SQSBatchResponse> Handler(SQSEvent ev, ILambdaContext context)
 	}
 	try
 	{
-		await linkIndexUpdater.SaveLinkIndex(linkIndex);
+		await linkIndexProvider.Save();
 		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);
 		return response;
 	}
 	catch (Exception ex)
 	{
+		// If we fail to update the link index, we need to return all messages to the queue
+		// so that they can be retried later.
 		context.Logger.LogError("Failed to update {bucketName}/{indexFile}. Returning all {recordCount} messages to the queue.", bucketName, indexFile, ev.Records.Count);
 		context.Logger.LogError(ex, ex.Message);
-		// If we fail to update the link index, we need to return all messages to the queue
 		var response = new SQSBatchResponse(ev.Records.Select(r => new SQSBatchResponse.BatchItemFailure
 		{
 			ItemIdentifier = r.MessageId
@@ -91,12 +92,12 @@ static LinkIndexEntry ConvertToLinkIndexEntry(S3EventNotification.S3EventNotific
 	SQSEvent.SQSMessage message, ILambdaContext context)
 {
 	var s3Event = S3EventNotification.ParseJson(message.Body);
-	var linkReferences = new ConcurrentBag<(S3EventNotification.S3EventNotificationRecord, LinkReference)>();
+	var recordLinkReferenceTuples = new ConcurrentBag<(S3EventNotification.S3EventNotificationRecord, LinkReference)>();
+	var linkReferenceProvider = new LinkReferenceProvider(s3Client, context.Logger, bucketName);
 	await Parallel.ForEachAsync(s3Event.Records, async (record, ctx) =>
 	{
-		var linkReferenceProvider = new LinkReferenceProvider(s3Client, context.Logger, record.S3.Bucket.Name);
 		var linkReference = await linkReferenceProvider.GetLinkReference(record.S3.Object.Key, ctx);
-		linkReferences.Add((record, linkReference));
+		recordLinkReferenceTuples.Add((record, linkReference));
 	});
-	return linkReferences;
+	return recordLinkReferenceTuples;
 }