fix: address scottgerring PR review feedback

jeastham1993 · claude · jeastham1993 · commit d38884411646 · 2026-03-05T15:24:32.000Z
- Fix wrong SubscribeOperation type in ServiceBusStickerPrintedWorker
  (was StickerClaimedEventV1, now StickerPrintedEventV1)
- Align envelope handling in ServiceBusStickerPrintedWorker to match
  ServiceBusStickerClaimedWorker (check ContentType, handle both
  CloudEvents and plain JSON)
- Register printJobs.completed.v1 Service Bus queue in Aspire setup
  so ServiceBusStickerPrintedWorker doesn't fail on Azure startup
- Fix activity tag in StickerPrintedHandler: stickerClaim.failed -&gt;
  stickerPrint.failed
- Add unit tests for StickerPrintedHandler covering null event,
  empty userId, user not found, and successful update paths
- Inject DatadogTransactionTracker into EventBridgeEventPublisher
  and call TrackTransactionAsync after publishing PrintJobQueuedEvent,
  matching the Kafka publisher behaviour
- Fix DLQ producer in KafkaStickerPrintedWorker: create once in
  StartAsync and reuse, rather than creating a new instance per call
- Remove duplicate StopAsync override in StickerPrintedWorker and
  StickerClaimedWorker that was calling messagingWorker.StopAsync
  twice (already called in ExecuteAsync finally block)
- Fix log level mismatches in LogMessages.cs: ReceivedMessage
  Critical-&gt;Information, GenericWarning Error-&gt;Warning
- Fix duplicate EventId 5 in LogMessages.cs: StoppingMessageProcessor
  reassigned to EventId 13
- Fix copy-pasta comment in integration test: "claimed sticker count"
  -&gt; "printed sticker count"
- Rename KafakStickerClaimedWorker -&gt; KafkaStickerClaimedWorker
  (filename and class name typo)

Co-Authored-By: Claude Sonnet 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/print-service/src/Stickerlandia.PrintService.AWS/EventBridgeEventPublisher.cs b/print-service/src/Stickerlandia.PrintService.AWS/EventBridgeEventPublisher.cs
@@ -31,7 +31,8 @@ namespace Stickerlandia.PrintService.AWS;
 public class EventBridgeEventPublisher(
     ILogger<EventBridgeEventPublisher> logger,
     IAmazonEventBridge client,
-    IOptions<AwsConfiguration> awsConfiguration) : IPrintServiceEventPublisher
+    IOptions<AwsConfiguration> awsConfiguration,
+    DatadogTransactionTracker transactionTracker) : IPrintServiceEventPublisher
 {
     [Channel("printJobs.queued.v1")]
     [PublishOperation(typeof(PrintJobQueuedEvent))]
@@ -49,6 +50,7 @@ public async Task PublishPrintJobQueuedEvent(PrintJobQueuedEvent printJobQueuedE
         };
 
         await PublishCloudEventAsync(cloudEvent);
+        await transactionTracker.TrackTransactionAsync(printJobQueuedEvent.PrintJobId, "print-job-queued");
     }
 
     [Channel("printJobs.failed.v1")]
diff --git a/print-service/src/Stickerlandia.PrintService.Core/Observability/LogMessages.cs b/print-service/src/Stickerlandia.PrintService.Core/Observability/LogMessages.cs
@@ -17,7 +17,7 @@ public static partial class Log
 
     [LoggerMessage(
         EventId = 0,
-        Level = LogLevel.Critical,
+        Level = LogLevel.Information,
         Message = "Received message from transport: {MessageTransport}")]
     public static partial void ReceivedMessage(
         ILogger logger, string messageTransport);
@@ -87,7 +87,7 @@ public static partial void UnknownException(
 
     [LoggerMessage(
         EventId = 10,
-        Level = LogLevel.Error,
+        Level = LogLevel.Warning,
         Message = "{Message}")]
     public static partial void GenericWarning(
         ILogger logger, string message, Exception? exception);
@@ -106,7 +106,7 @@ public static partial void TransactionTrackingFailed(
         ILogger logger, string transactionId, string checkpoint, Exception exception);
 
     [LoggerMessage(
-        EventId = 5,
+        EventId = 13,
         Level = LogLevel.Trace,
         Message = "Stopping message processor for transport: {MessageTransport}")]
     public static partial void StoppingMessageProcessor(
diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerClaimedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerClaimedWorker.cs
@@ -20,8 +20,8 @@
 namespace Stickerlandia.UserManagement.Agnostic;
 
 [AsyncApi]
-public class KafakStickerClaimedWorker(
-    ILogger<KafakStickerClaimedWorker> logger,
+public class KafkaStickerClaimedWorker(
+    ILogger<KafkaStickerClaimedWorker> logger,
     IServiceScopeFactory serviceScopeFactory,
     ConsumerConfig consumerConfig,
     ProducerConfig producerConfig,
diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerPrintedWorker.cs
@@ -33,6 +33,7 @@ public class KafkaStickerPrintedWorker(
     private const string dlqTopic = "printJobs.completed.v1.dlq";
 
     private IConsumer<string, string>? _consumer;
+    private IProducer<string, string>? _dlqProducer;
 
     [Channel("printJobs.completed.v1")]
     [SubscribeOperation(typeof(StickerPrintedEventV1))]
@@ -67,9 +68,7 @@ private async Task<bool> ProcessMessageAsync(StickerPrintedHandler handler,
 
     private void SendToDLQ(ConsumeResult<string, string> consumeResult, StickerPrintedEventV1 evtData)
     {
-        using var producer = new ProducerBuilder<string, string>(producerConfig).Build();
-
-        producer.Produce(dlqTopic,
+        _dlqProducer!.Produce(dlqTopic,
             new Message<string, string> { Key = evtData.UserId, Value = consumeResult.Message.Value },
             (deliveryReport) =>
             {
@@ -78,8 +77,6 @@ private void SendToDLQ(ConsumeResult<string, string> consumeResult, StickerPrint
                 else
                     Log.MessageSentToDlq(logger, "", null);
             });
-
-        producer.Flush(TimeSpan.FromSeconds(10));
     }
 
     public Task StartAsync()
@@ -89,6 +86,8 @@ public Task StartAsync()
         _consumer = consumerFactory.Create(consumerConfig);
         _consumer.Subscribe(topic);
 
+        _dlqProducer = new ProducerBuilder<string, string>(producerConfig).Build();
+
         return Task.CompletedTask;
     }
 
@@ -130,6 +129,11 @@ public Task StopAsync(CancellationToken cancellationToken)
         _consumer?.Close();
         _consumer?.Dispose();
         _consumer = null;
+
+        _dlqProducer?.Flush(TimeSpan.FromSeconds(10));
+        _dlqProducer?.Dispose();
+        _dlqProducer = null;
+
         return Task.CompletedTask;
     }
 }
diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/ServiceExtensions.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/ServiceExtensions.cs
@@ -92,7 +92,7 @@ public static IServiceCollection AddKafkaMessaging(this IServiceCollection servi
 
         // Register event publisher as singleton
         services.AddSingleton<IUserEventPublisher, KafkaEventPublisher>();
-        services.AddKeyedSingleton<IMessagingWorker, KafakStickerClaimedWorker>("stickerClaimed");
+        services.AddKeyedSingleton<IMessagingWorker, KafkaStickerClaimedWorker>("stickerClaimed");
         services.AddKeyedSingleton<IMessagingWorker, KafkaStickerPrintedWorker>("stickerPrinted");
 
         return services;
diff --git a/user-management/src/Stickerlandia.UserManagement.Aspire/AppBuilderExtensions.cs b/user-management/src/Stickerlandia.UserManagement.Aspire/AppBuilderExtensions.cs
@@ -253,6 +253,9 @@ public static InfrastructureResources WithAzureNativeServices(this IDistributedA
             .AddServiceBusQueue("users-stickerClaimed-v1", "users.stickerClaimed.v1")
             .WithServiceBusTestCommands();
 
+        serviceBus
+            .AddServiceBusQueue("printJobs-completed-v1", "printJobs.completed.v1");
+
         var topic = serviceBus
             .AddServiceBusTopic("users-userRegistered-v1", "users.userRegistered.v1");
         topic.AddServiceBusSubscription("noop");
diff --git a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerPrintedWorker.cs
@@ -8,15 +8,13 @@
 // This product includes software developed at Datadog (https://www.datadoghq.com/).
 // Copyright 2025 Datadog, Inc.
 
+using System.Text.Json;
 using Azure.Messaging.ServiceBus;
-using CloudNative.CloudEvents;
-using CloudNative.CloudEvents.SystemTextJson;
 using Datadog.Trace;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Saunter.Attributes;
 using Stickerlandia.UserManagement.Core;
-using Stickerlandia.UserManagement.Core.StickerClaimedEvent;
 using Stickerlandia.UserManagement.Core.StickerPrintedEvent;
 using Log = Stickerlandia.UserManagement.Core.Observability.Log;
 
@@ -46,7 +44,7 @@ public ServiceBusStickerPrintedWorker(ILogger<ServiceBusStickerPrintedWorker> lo
     }
 
     [Channel("printJobs.completed.v1")]
-    [SubscribeOperation(typeof(StickerClaimedEventV1))]
+    [SubscribeOperation(typeof(StickerPrintedEventV1))]
     private async Task ProcessMessageAsync(ProcessMessageEventArgs args)
     {
         using var processSpan = Tracer.Instance.StartActive($"process printJobs.completed.v1");
@@ -57,11 +55,18 @@ private async Task ProcessMessageAsync(ProcessMessageEventArgs args)
         var messageBody = args.Message.Body.ToString();
         Log.ReceivedMessage(_logger, "ServiceBus");
 
-        var detailBytes = System.Text.Encoding.UTF8.GetBytes(messageBody);
-        var formatter = new JsonEventFormatter<StickerPrintedEventV1>();
-        var cloudEvent = await formatter.DecodeStructuredModeMessageAsync(
-            new MemoryStream(detailBytes), null, new List<CloudEventAttribute>());
-        var stickerPrinted = (StickerPrintedEventV1?)cloudEvent.Data;
+        StickerPrintedEventV1? stickerPrinted;
+        if (string.Equals(args.Message.ContentType, "application/cloudevents+json", StringComparison.OrdinalIgnoreCase))
+        {
+            using var document = JsonDocument.Parse(messageBody);
+            stickerPrinted = document.RootElement.TryGetProperty("data", out var dataElement)
+                ? JsonSerializer.Deserialize<StickerPrintedEventV1>(dataElement.GetRawText())
+                : null;
+        }
+        else
+        {
+            stickerPrinted = JsonSerializer.Deserialize<StickerPrintedEventV1>(messageBody);
+        }
 
         if (stickerPrinted == null) await args.DeadLetterMessageAsync(args.Message, "Message body cannot be deserialized");
         
diff --git a/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedHandler.cs b/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedHandler.cs
@@ -37,7 +37,7 @@ public async Task Handle(StickerPrintedEventV1 eventV1)
         }
         catch (Exception ex)
         {
-            Activity.Current?.AddTag("stickerClaim.failed", true);
+            Activity.Current?.AddTag("stickerPrint.failed", true);
             Activity.Current?.AddTag("error.message", ex.Message);
 
             throw;
diff --git a/user-management/src/Stickerlandia.UserManagement.Worker/StickerClaimedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Worker/StickerClaimedWorker.cs
@@ -61,10 +61,4 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             await messagingWorker.StopAsync(stoppingToken);
         }
     }
-    
-    public override async Task StopAsync(CancellationToken cancellationToken)
-    {
-        await messagingWorker.StopAsync(cancellationToken);
-        await base.StopAsync(cancellationToken);
-    }
 }
diff --git a/user-management/src/Stickerlandia.UserManagement.Worker/StickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Worker/StickerPrintedWorker.cs
@@ -61,10 +61,4 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             await messagingWorker.StopAsync(stoppingToken);
         }
     }
-    
-    public override async Task StopAsync(CancellationToken cancellationToken)
-    {
-        await messagingWorker.StopAsync(cancellationToken);
-        await base.StopAsync(cancellationToken);
-    }
 }
diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/AccountTests.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/AccountTests.cs
@@ -95,7 +95,7 @@ public async Task WhenStickerIsPrintedThenAUsersPrintedCountShouldIncrement()
             
             var user = await _driver.GetUserAccount(loginResponse.AuthToken);
 
-            // Expect the claimed sticker count to be 1, break after completed.
+            // Expect the printed sticker count to be 1, break after completed.
             if (user!.PrintedStickerCount == 1)
             {
                 break;
diff --git a/user-management/tests/Stickerlandia.UserManagement.UnitTest/KafkaWorkerLifecycleTests.cs b/user-management/tests/Stickerlandia.UserManagement.UnitTest/KafkaWorkerLifecycleTests.cs
@@ -80,20 +80,20 @@ public async Task PollAsync_WhenNoMessage_DoesNotCommit()
     }
 }
 
-public class KafakStickerClaimedWorkerTests
+public class KafkaStickerClaimedWorkerTests
 {
     private readonly IConsumer<string, string> _consumer;
     private readonly IKafkaConsumerFactory _factory;
-    private readonly KafakStickerClaimedWorker _worker;
+    private readonly KafkaStickerClaimedWorker _worker;
 
-    public KafakStickerClaimedWorkerTests()
+    public KafkaStickerClaimedWorkerTests()
     {
         _consumer = A.Fake<IConsumer<string, string>>();
         _factory = A.Fake<IKafkaConsumerFactory>();
         A.CallTo(() => _factory.Create(A<ConsumerConfig>.Ignored)).Returns(_consumer);
 
-        _worker = new KafakStickerClaimedWorker(
-            A.Fake<ILogger<KafakStickerClaimedWorker>>(),
+        _worker = new KafkaStickerClaimedWorker(
+            A.Fake<ILogger<KafkaStickerClaimedWorker>>(),
             A.Fake<IServiceScopeFactory>(),
             new ConsumerConfig { BootstrapServers = "localhost:9092", GroupId = "test" },
             new ProducerConfig { BootstrapServers = "localhost:9092" },
diff --git a/user-management/tests/Stickerlandia.UserManagement.UnitTest/StickerPrintedHandlerTests.cs b/user-management/tests/Stickerlandia.UserManagement.UnitTest/StickerPrintedHandlerTests.cs
@@ -0,0 +1,99 @@
+/*
+ * Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+ * This product includes software developed at Datadog (https://www.datadoghq.com/).
+ * Copyright 2025-Present Datadog, Inc.
+ */
+
+using System.Diagnostics.Metrics;
+using Microsoft.AspNetCore.Identity;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Stickerlandia.UserManagement.Core;
+using Stickerlandia.UserManagement.Core.StickerPrintedEvent;
+
+namespace Stickerlandia.UserManagement.UnitTest;
+
+public class StickerPrintedHandlerTests
+{
+    [Fact]
+    public async Task GivenNullEvent_ShouldThrowArgumentException()
+    {
+        using var userManager = CreateFakeUserManager();
+        var handler = new StickerPrintedHandler(userManager);
+
+        await Assert.ThrowsAsync<ArgumentException>(() => handler.Handle(null!));
+    }
+
+    [Fact]
+    public async Task GivenEventWithEmptyUserId_ShouldThrowArgumentException()
+    {
+        using var userManager = CreateFakeUserManager();
+        var handler = new StickerPrintedHandler(userManager);
+
+        await Assert.ThrowsAsync<ArgumentException>(() =>
+            handler.Handle(new StickerPrintedEventV1 { UserId = "" }));
+    }
+
+    [Fact]
+    public async Task GivenUserDoesNotExist_ShouldThrowInvalidUserException()
+    {
+        using var userManager = CreateFakeUserManager();
+        A.CallTo(() => userManager.FindByIdAsync(A<string>.Ignored))
+            .Returns(Task.FromResult<PostgresUserAccount?>(null));
+
+        var handler = new StickerPrintedHandler(userManager);
+
+        await Assert.ThrowsAsync<InvalidUserException>(() =>
+            handler.Handle(new StickerPrintedEventV1 { UserId = "nonexistent-id" }));
+    }
+
+    [Fact]
+    public async Task GivenValidEvent_ShouldIncrementPrintedStickerCount()
+    {
+        var userId = "valid-user-id";
+        var account = new PostgresUserAccount { Id = userId };
+        var initialCount = account.PrintedStickerCount;
+
+        PostgresUserAccount? updatedAccount = null;
+
+        using var userManager = CreateFakeUserManager();
+        A.CallTo(() => userManager.FindByIdAsync(userId))
+            .Returns(Task.FromResult<PostgresUserAccount?>(account));
+        A.CallTo(() => userManager.UpdateAsync(A<PostgresUserAccount>.Ignored))
+            .Invokes((PostgresUserAccount a) => updatedAccount = a)
+            .Returns(Task.FromResult(IdentityResult.Success));
+
+        var handler = new StickerPrintedHandler(userManager);
+
+        await handler.Handle(new StickerPrintedEventV1 { UserId = userId });
+
+        updatedAccount.Should().NotBeNull();
+        updatedAccount!.PrintedStickerCount.Should().Be(initialCount + 1);
+    }
+
+    private static UserManager<PostgresUserAccount> CreateFakeUserManager()
+    {
+        var userManager = A.Fake<UserManager<PostgresUserAccount>>(options =>
+            options.WithArgumentsForConstructor(() => new UserManager<PostgresUserAccount>(
+                A.Fake<IUserStore<PostgresUserAccount>>(),
+                A.Fake<IOptions<IdentityOptions>>(),
+                A.Fake<IPasswordHasher<PostgresUserAccount>>(),
+                new List<IUserValidator<PostgresUserAccount>>(),
+                new List<IPasswordValidator<PostgresUserAccount>>(),
+                A.Fake<ILookupNormalizer>(),
+                A.Fake<IdentityErrorDescriber>(),
+                CreateFakeServiceProvider(),
+                A.Fake<ILogger<UserManager<PostgresUserAccount>>>()
+            )));
+
+        return userManager;
+    }
+
+    private static IServiceProvider CreateFakeServiceProvider()
+    {
+        var serviceProvider = A.Fake<IServiceProvider>();
+        A.CallTo(() => serviceProvider.GetService(typeof(IMeterFactory)))
+            .Returns(null);
+        return serviceProvider;
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ public async Task Handle(StickerPrintedEventV1 eventV1)`
`37`	`37`	`}`
`38`	`38`	`catch (Exception ex)`
`39`	`39`	`{`
`40`		`- Activity.Current?.AddTag("stickerClaim.failed", true);`
	`40`	`+ Activity.Current?.AddTag("stickerPrint.failed", true);`
`41`	`41`	`Activity.Current?.AddTag("error.message", ex.Message);`
`42`	`42`
`43`	`43`	`throw;`
Original file line number	Diff line number	Diff line change
`@@ -61,10 +61,4 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)`
`61`	`61`	`await messagingWorker.StopAsync(stoppingToken);`
`62`	`62`	`}`
`63`	`63`	`}`
`64`		`-`
`65`		`- public override async Task StopAsync(CancellationToken cancellationToken)`
`66`		`- {`
`67`		`- await messagingWorker.StopAsync(cancellationToken);`
`68`		`- await base.StopAsync(cancellationToken);`
`69`		`- }`
`70`	`64`	`}`