diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml deleted file mode 100644 index 3c68ee90..00000000 --- a/.github/workflows/claude-code-review.yml +++ /dev/null @@ -1,44 +0,0 @@ -name: Claude Code Review - -on: - pull_request: - types: [opened, synchronize, ready_for_review, reopened] - # Optional: Only run on specific file changes - # paths: - # - "src/**/*.ts" - # - "src/**/*.tsx" - # - "src/**/*.js" - # - "src/**/*.jsx" - -jobs: - claude-review: - # Optional: Filter by PR author - # if: | - # github.event.pull_request.user.login == 'external-contributor' || - # github.event.pull_request.user.login == 'new-developer' || - # github.event.pull_request.author_association == 'FIRST_TIME_CONTRIBUTOR' - - runs-on: ubuntu-latest - permissions: - contents: read - pull-requests: read - issues: read - id-token: write - - steps: - - name: Checkout repository - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 - with: - fetch-depth: 1 - - - name: Run Claude Code Review - id: claude-review - uses: anthropics/claude-code-action@cd77b50d2b0808657f8e6774085c8bf54484351c # v1.0.72 - with: - anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} - plugin_marketplaces: 'https://github.com/anthropics/claude-code.git' - plugins: 'code-review@claude-code-plugins' - prompt: '/code-review:code-review ${{ github.repository }}/pull/${{ github.event.pull_request.number }}' - # See https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md - # or https://code.claude.com/docs/en/cli-reference for available options - diff --git a/.github/workflows/claude-code-review.yml.disabled b/.github/workflows/claude-code-review.yml.disabled deleted file mode 100644 index 82e06e2e..00000000 --- a/.github/workflows/claude-code-review.yml.disabled +++ /dev/null @@ -1,78 +0,0 @@ -name: Claude Code Review - -on: - pull_request: - types: [opened, synchronize] - # Optional: Only run on specific file changes - # paths: - # - "src/**/*.ts" - # - "src/**/*.tsx" - # - "src/**/*.js" - # - "src/**/*.jsx" - -jobs: - claude-review: - # Optional: Filter by PR author - # if: | - # github.event.pull_request.user.login == 'external-contributor' || - # github.event.pull_request.user.login == 'new-developer' || - # github.event.pull_request.author_association == 'FIRST_TIME_CONTRIBUTOR' - - runs-on: ubuntu-latest - permissions: - contents: read - pull-requests: read - issues: read - id-token: write - - steps: - - name: Checkout repository - uses: actions/checkout@v4 - with: - fetch-depth: 1 - - - name: Run Claude Code Review - id: claude-review - uses: anthropics/claude-code-action@beta - with: - anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} - - # Optional: Specify model (defaults to Claude Sonnet 4, uncomment for Claude Opus 4) - # model: "claude-opus-4-20250514" - - # Direct prompt for automated review (no @claude mention needed) - direct_prompt: | - Please review this pull request and provide feedback on: - - Code quality and best practices - - Potential bugs or issues - - Performance considerations - - Security concerns - - Test coverage - - Be constructive and helpful in your feedback. - - # Optional: Use sticky comments to make Claude reuse the same comment on subsequent pushes to the same PR - # use_sticky_comment: true - - # Optional: Customize review based on file types - # direct_prompt: | - # Review this PR focusing on: - # - For TypeScript files: Type safety and proper interface usage - # - For API endpoints: Security, input validation, and error handling - # - For React components: Performance, accessibility, and best practices - # - For tests: Coverage, edge cases, and test quality - - # Optional: Different prompts for different authors - # direct_prompt: | - # ${{ github.event.pull_request.author_association == 'FIRST_TIME_CONTRIBUTOR' && - # 'Welcome! Please review this PR from a first-time contributor. Be encouraging and provide detailed explanations for any suggestions.' || - # 'Please provide a thorough code review focusing on our coding standards and best practices.' }} - - # Optional: Add specific tools for running tests or linting - # allowed_tools: "Bash(npm run test),Bash(npm run lint),Bash(npm run typecheck)" - - # Optional: Skip review for certain conditions - # if: | - # !contains(github.event.pull_request.title, '[skip-review]') && - # !contains(github.event.pull_request.title, '[WIP]') - diff --git a/docker-compose.yml b/docker-compose.yml index 25bb74e9..349ab4a6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -379,11 +379,14 @@ services: condition: service_healthy environment: # DataDog APM configuration + DD_API_KEY: ${DD_API_KEY} + DD_SITE: ${DD_SITE:-datadoghq.com} DD_SERVICE: print-service DD_ENV: ${ENV:-development} DD_VERSION: ${COMMIT_SHA:-latest} DD_AGENT_HOST: datadog-agent DD_DATA_STREAMS_ENABLED: "true" + DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED: "true" DD_TRACE_OTEL_ENABLED: "true" DD_LOGS_INJECTION: "true" DD_RUNTIME_METRICS_ENABLED: "true" @@ -433,11 +436,14 @@ services: condition: service_healthy environment: # DataDog APM configuration + DD_API_KEY: ${DD_API_KEY} + DD_SITE: ${DD_SITE:-datadoghq.com} DD_SERVICE: print-service DD_ENV: ${ENV:-development} DD_VERSION: ${COMMIT_SHA:-latest} DD_AGENT_HOST: datadog-agent DD_DATA_STREAMS_ENABLED: "true" + DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED: "true" DD_TRACE_OTEL_ENABLED: "true" DD_LOGS_INJECTION: "true" DD_RUNTIME_METRICS_ENABLED: "true" @@ -474,11 +480,14 @@ services: - "8086:8080" environment: # DataDog APM configuration + DD_API_KEY: ${DD_API_KEY} + DD_SITE: ${DD_SITE:-datadoghq.com} DD_SERVICE: print-service-client DD_ENV: ${ENV:-development} DD_VERSION: ${COMMIT_SHA:-latest} DD_AGENT_HOST: datadog-agent DD_DATA_STREAMS_ENABLED: "true" + DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED: "true" DD_TRACE_OTEL_ENABLED: "true" DD_LOGS_INJECTION: "true" DD_RUNTIME_METRICS_ENABLED: "true" @@ -538,7 +547,8 @@ services: DD_ENV: ${ENV:-development} DD_VERSION: ${COMMIT_SHA:-latest} DD_AGENT_HOST: datadog-agent - DD_DATA_STREAMS_ENABLED: true + DD_DATA_STREAMS_ENABLED: "true" + DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED: "true" DD_TRACE_OTEL_ENABLED: "true" DD_LOGS_INJECTION: "true" DD_RUNTIME_METRICS_ENABLED: "true" @@ -592,7 +602,8 @@ services: DD_ENV: ${ENV:-development} DD_VERSION: ${COMMIT_SHA:-latest} DD_AGENT_HOST: datadog-agent - DD_DATA_STREAMS_ENABLED: true + DD_DATA_STREAMS_ENABLED: "true" + DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED: "true" DD_TRACE_OTEL_ENABLED: "true" DD_LOGS_INJECTION: "true" DD_RUNTIME_METRICS_ENABLED: "true" @@ -716,7 +727,8 @@ services: DD_ENV: ${ENV:-development} DD_VERSION: ${COMMIT_SHA:-latest} DD_AGENT_HOST: datadog-agent - DD_DATA_STREAMS_ENABLED: true + DD_DATA_STREAMS_ENABLED: "true" + DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED: "true" # Server configuration SERVER_PORT: 8080 @@ -768,6 +780,7 @@ services: # Load testing service - runs k6 load tests against the stack load-simulator: image: grafana/k6:latest + restart: always volumes: - ./load-tests:/scripts environment: diff --git a/mise.toml b/mise.toml index 855e6ea0..f55bded6 100644 --- a/mise.toml +++ b/mise.toml @@ -280,3 +280,19 @@ docker compose --profile load-test run --rm load-simulator docker compose --profile load-test down -v """ +[tasks."aws:load-test:smoke"] +description = "Run load tests against AWS deployment" +dir = "e2e" +depends = ["ui-test:install"] +run = """ +BASE_URL=$(aws cloudformation describe-stacks \ + --stack-name "StickerlandiaSharedResources-${ENV:-dev}" \ + --query "Stacks[0].Outputs[?OutputKey=='BaseUrl'].OutputValue" \ + --output text) +DEPLOYMENT_HOST_URL=$(aws cloudformation describe-stacks \ + --stack-name "StickerlandiaSharedResources-${ENV:-dev}" \ + --query "Stacks[0].Outputs[?OutputKey=='BaseUrl'].OutputValue" \ + --output text) +echo "Running tests against: $BASE_URL" +BASE_URL=$BASE_URL DEPLOYMENT_HOST_URL=$DEPLOYMENT_HOST_URL npx playwright test +""" \ No newline at end of file diff --git a/print-service/src/Stickerlandia.PrintService.AWS/EventBridgeEventPublisher.cs b/print-service/src/Stickerlandia.PrintService.AWS/EventBridgeEventPublisher.cs index 6a846031..6ce621af 100644 --- a/print-service/src/Stickerlandia.PrintService.AWS/EventBridgeEventPublisher.cs +++ b/print-service/src/Stickerlandia.PrintService.AWS/EventBridgeEventPublisher.cs @@ -9,10 +9,12 @@ // Copyright 2025 Datadog, Inc. using System.Diagnostics; +using System.Text.Json.Nodes; using Amazon.EventBridge; using Amazon.EventBridge.Model; using CloudNative.CloudEvents; using CloudNative.CloudEvents.SystemTextJson; +using Datadog.Trace; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Saunter.Attributes; @@ -29,7 +31,8 @@ namespace Stickerlandia.PrintService.AWS; public class EventBridgeEventPublisher( ILogger logger, IAmazonEventBridge client, - IOptions awsConfiguration) : IPrintServiceEventPublisher + IOptions awsConfiguration, + IDatadogTransactionTracker transactionTracker) : IPrintServiceEventPublisher { [Channel("printJobs.queued.v1")] [PublishOperation(typeof(PrintJobQueuedEvent))] @@ -47,6 +50,7 @@ public async Task PublishPrintJobQueuedEvent(PrintJobQueuedEvent printJobQueuedE }; await PublishCloudEventAsync(cloudEvent); + await transactionTracker.TrackTransactionAsync(printJobQueuedEvent.PrintJobId, "print-job-queued"); } [Channel("printJobs.failed.v1")] @@ -139,6 +143,18 @@ internal async Task PublishCloudEventAsync(CloudEvent cloudEvent) var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); var jsonString = System.Text.Encoding.UTF8.GetString(data.Span); + if (Tracer.Instance.ActiveScope is not null) + { + var jsonNode = JsonNode.Parse(jsonString)!; + new SpanContextInjector().InjectIncludingDsm(jsonNode, SetHeader, Tracer.Instance.ActiveScope.Span.Context, "eventbridge", cloudEvent.Type!); + jsonString = jsonNode.ToJsonString(); + } + else + { + Log.GenericWarning(logger, "DSM injection skipped for EventBridge: no active Datadog scope", null); + activity?.SetTag("dsm.injection_skipped", "no active scope"); + } + var response = await client.PutEventsAsync(new PutEventsRequest() { Entries = new List(1) @@ -146,7 +162,7 @@ internal async Task PublishCloudEventAsync(CloudEvent cloudEvent) new() { EventBusName = awsConfiguration.Value.EventBusName, - Source = $"{Environment.GetEnvironmentVariable("ENV") ?? "dev"}.users", + Source = $"{Environment.GetEnvironmentVariable("ENV") ?? "dev"}.print", DetailType = cloudEvent.Type, Detail = jsonString, } @@ -171,4 +187,11 @@ internal async Task PublishCloudEventAsync(CloudEvent cloudEvent) throw; } } + + private static void SetHeader(JsonNode jsonNode, string key, string value) + { + if (jsonNode["_datadog"] == null) jsonNode["_datadog"] = new JsonObject(); + + jsonNode["_datadog"]![key] = value; + } } \ No newline at end of file diff --git a/print-service/src/Stickerlandia.PrintService.AWS/ServiceExtensions.cs b/print-service/src/Stickerlandia.PrintService.AWS/ServiceExtensions.cs index 1bf977b1..c04ad8b2 100644 --- a/print-service/src/Stickerlandia.PrintService.AWS/ServiceExtensions.cs +++ b/print-service/src/Stickerlandia.PrintService.AWS/ServiceExtensions.cs @@ -16,6 +16,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Stickerlandia.PrintService.Core; +using Stickerlandia.PrintService.Core.Observability; using Stickerlandia.PrintService.Core.Outbox; using Stickerlandia.PrintService.Core.PrintJobs; @@ -61,6 +62,9 @@ public static IServiceCollection AddAwsAdapters(this IServiceCollection services services.AddSingleton(); + services.AddHttpClient(); + services.AddSingleton(); + return services; } } \ No newline at end of file diff --git a/print-service/src/Stickerlandia.PrintService.Agnostic/KafkaEventPublisher.cs b/print-service/src/Stickerlandia.PrintService.Agnostic/KafkaEventPublisher.cs index d20917ef..5a04ed43 100644 --- a/print-service/src/Stickerlandia.PrintService.Agnostic/KafkaEventPublisher.cs +++ b/print-service/src/Stickerlandia.PrintService.Agnostic/KafkaEventPublisher.cs @@ -7,6 +7,7 @@ using CloudNative.CloudEvents; using CloudNative.CloudEvents.SystemTextJson; using Confluent.Kafka; +using Datadog.Trace; using Microsoft.Extensions.Logging; using Saunter.Attributes; using Stickerlandia.PrintService.Core; @@ -17,7 +18,7 @@ namespace Stickerlandia.PrintService.Agnostic; -public class KafkaEventPublisher(ProducerConfig config, ILogger logger) : IPrintServiceEventPublisher +public class KafkaEventPublisher(ProducerConfig config, ILogger logger, IDatadogTransactionTracker transactionTracker) : IPrintServiceEventPublisher { [Channel("printJobs.queued.v1")] [PublishOperation(typeof(PrintJobQueuedEvent))] @@ -34,7 +35,8 @@ public async Task PublishPrintJobQueuedEvent(PrintJobQueuedEvent printJobQueuedE Data = printJobQueuedEvent }; - await Publish(cloudEvent); + await Publish(cloudEvent); + await transactionTracker.TrackTransactionAsync(printJobQueuedEvent.PrintJobId, "print-job-queued"); } [Channel("printJobs.failed.v1")] @@ -52,7 +54,7 @@ public async Task PublishPrintJobFailedEvent(PrintJobFailedEvent printJobFailedE Data = printJobFailedEvent }; - await Publish(cloudEvent); + await Publish(cloudEvent); } [Channel("print.printerRegistered.v1")] @@ -126,13 +128,33 @@ private async Task Publish(CloudEvent cloudEvent) var formatter = new JsonEventFormatter(); var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + var message = new Message + { + Key = cloudEvent.Id!, + Value = Encoding.UTF8.GetString(data.Span) + }; + + if (Tracer.Instance.ActiveScope is not null) + { + message.Headers = new Headers(); + new SpanContextInjector().InjectIncludingDsm( + message.Headers, + (headers, key, value) => headers.Add(key, Encoding.UTF8.GetBytes(value)), + Tracer.Instance.ActiveScope.Span.Context, + "kafka", + cloudEvent.Type!); + } + else + { + Log.GenericWarning(logger, "DSM injection skipped for Kafka: no active Datadog scope", null); + activity?.SetTag("dsm.injection_skipped", "no active scope"); + } + using var producer = new ProducerBuilder(config).Build(); try { - var deliveryReport = await producer.ProduceAsync(cloudEvent.Type, - new Message { Key = cloudEvent.Id!, Value = Encoding.UTF8.GetString(data.Span) }, - default); + var deliveryReport = await producer.ProduceAsync(cloudEvent.Type, message, default); if (deliveryReport.Status == PersistenceStatus.PossiblyPersisted) { diff --git a/print-service/src/Stickerlandia.PrintService.Agnostic/ServiceExtensions.cs b/print-service/src/Stickerlandia.PrintService.Agnostic/ServiceExtensions.cs index e78a7869..b0692b45 100644 --- a/print-service/src/Stickerlandia.PrintService.Agnostic/ServiceExtensions.cs +++ b/print-service/src/Stickerlandia.PrintService.Agnostic/ServiceExtensions.cs @@ -11,6 +11,7 @@ using Stickerlandia.PrintService.Agnostic.Data; using Stickerlandia.PrintService.Agnostic.Repositories; using Stickerlandia.PrintService.Core; +using Stickerlandia.PrintService.Core.Observability; using Stickerlandia.PrintService.Core.Outbox; using Stickerlandia.PrintService.Core.PrintJobs; @@ -122,6 +123,9 @@ public static IServiceCollection AddKafkaMessaging(this IServiceCollection servi services.AddSingleton(producerConfig); services.AddSingleton(consumerConfig); + services.AddHttpClient(); + services.AddSingleton(); + // Register event publisher as singleton services.AddSingleton(); diff --git a/print-service/src/Stickerlandia.PrintService.Api/Configurations/AuthenticationExtensions.cs b/print-service/src/Stickerlandia.PrintService.Api/Configurations/AuthenticationExtensions.cs index 040df728..a91165f3 100644 --- a/print-service/src/Stickerlandia.PrintService.Api/Configurations/AuthenticationExtensions.cs +++ b/print-service/src/Stickerlandia.PrintService.Api/Configurations/AuthenticationExtensions.cs @@ -10,7 +10,7 @@ using Microsoft.IdentityModel.Protocols.OpenIdConnect; using Microsoft.IdentityModel.Tokens; -#pragma warning disable CA5400 +#pragma warning disable CA5400, CA2000 namespace Stickerlandia.PrintService.Api.Configurations; diff --git a/print-service/src/Stickerlandia.PrintService.Client/Program.cs b/print-service/src/Stickerlandia.PrintService.Client/Program.cs index 50d85553..ce3d816b 100644 --- a/print-service/src/Stickerlandia.PrintService.Client/Program.cs +++ b/print-service/src/Stickerlandia.PrintService.Client/Program.cs @@ -15,6 +15,7 @@ using Stickerlandia.PrintService.Client.Configuration; using Stickerlandia.PrintService.Client.Services; using Stickerlandia.PrintService.Client.Telemetry; +using CoreObs = Stickerlandia.PrintService.Core.Observability; var builder = WebApplication.CreateBuilder(args); builder.Configuration.AddEnvironmentVariables(); @@ -77,6 +78,7 @@ builder.Services.AddRazorComponents() .AddInteractiveServerComponents(); +builder.Services.AddSingleton(); // Add configuration service (singleton - shared across app) builder.Services.AddSingleton(); diff --git a/print-service/src/Stickerlandia.PrintService.Client/Services/PrintJobPollingService.cs b/print-service/src/Stickerlandia.PrintService.Client/Services/PrintJobPollingService.cs index 262c9901..d2071b7b 100644 --- a/print-service/src/Stickerlandia.PrintService.Client/Services/PrintJobPollingService.cs +++ b/print-service/src/Stickerlandia.PrintService.Client/Services/PrintJobPollingService.cs @@ -11,6 +11,7 @@ using Stickerlandia.PrintService.Client.Configuration; using Stickerlandia.PrintService.Client.Models; using Stickerlandia.PrintService.Client.Telemetry; +using CoreObs = Stickerlandia.PrintService.Core.Observability; namespace Stickerlandia.PrintService.Client.Services; @@ -25,6 +26,7 @@ internal sealed class PrintJobPollingService : BackgroundService private readonly ClientStatusService _statusService; private readonly PrintClientInstrumentation _instrumentation; private readonly ILogger _logger; + private readonly CoreObs.IDatadogTransactionTracker _transactionTracker; public PrintJobPollingService( IPrintServiceApiClient apiClient, @@ -32,7 +34,7 @@ public PrintJobPollingService( IConfigurationService configService, ClientStatusService statusService, PrintClientInstrumentation instrumentation, - ILogger logger) + ILogger logger, CoreObs.IDatadogTransactionTracker transactionTracker) { _apiClient = apiClient; _localStorage = localStorage; @@ -40,6 +42,7 @@ public PrintJobPollingService( _statusService = statusService; _instrumentation = instrumentation; _logger = logger; + _transactionTracker = transactionTracker; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -130,7 +133,21 @@ private async Task ProcessJobAsync(PrintJobDto job) { _logger.LogInformation("Processing job {JobId} for sticker {StickerId}", job.PrintJobId, job.StickerId); + // Extract DSM context and activate the Datadog scope before starting the OTel activity, + // so the activity is parented to the DSM-extracted context rather than the ambient poll span. + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + job, + GetHeader, + "queue", + "print_queue"); + + using var scope = Tracer.Instance.StartActive( + "process print.job", + new SpanCreationSettings { Parent = extractedContext }); + using var activity = PrintClientInstrumentation.StartProcessJobActivity(job); + + await _transactionTracker.TrackTransactionAsync(job.PrintJobId, "print-received"); var stopwatch = Stopwatch.StartNew(); try @@ -159,20 +176,7 @@ private async Task ProcessJobAsync(PrintJobDto job) _logger.LogInformation("Job {JobId} processed and acknowledged", job.PrintJobId); ackActivity?.SetStatus(ActivityStatusCode.Ok); _instrumentation.RecordAcknowledgementSucceeded(); - - // Mark consumption as complete - var propagator = new SpanContextExtractor(); - var extractedContext = propagator.ExtractIncludingDsm( - job, - GetHeader, - "queue", - "print_queue"); - using var scope = Tracer.Instance.StartActive( - "processed print.job", - new SpanCreationSettings - { - Parent = extractedContext - }); + await _transactionTracker.TrackTransactionAsync(job.PrintJobId, "print-completed"); } else { diff --git a/print-service/src/Stickerlandia.PrintService.Client/Stickerlandia.PrintService.Client.csproj b/print-service/src/Stickerlandia.PrintService.Client/Stickerlandia.PrintService.Client.csproj index ec2e7f18..39ef21ea 100644 --- a/print-service/src/Stickerlandia.PrintService.Client/Stickerlandia.PrintService.Client.csproj +++ b/print-service/src/Stickerlandia.PrintService.Client/Stickerlandia.PrintService.Client.csproj @@ -7,6 +7,10 @@ true + + + + @@ -36,4 +40,17 @@ + + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\415ea02f-bc10-4ef6-89e7-2f769009f87e.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\506dccea-4c39-4e96-95bf-6858e30d043e.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\5270f1d5-9e9a-45b7-91dc-9eb65e4b23cb.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\6868ccee-3ddf-4c36-813a-e51c130865c3.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\82a74161-0eca-4fe1-a3d5-bb3fd86f2afc.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\c1f03d74-0c92-429f-807c-e8e86dffa326.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\c64e482c-9200-4406-8501-56a17bc8ac7d.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\c72499f4-ce72-4169-be33-7b3eb918f614.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\d8e56c15-6e4c-4cb6-b9e4-3a277c0725b0.json" /> + <_ContentIncludedByDefault Remove="print-jobs\2026-03-24\e03b614b-b23d-4ebb-a747-bfdaee9f67ee.json" /> + + diff --git a/print-service/src/Stickerlandia.PrintService.Core/Observability/DatadogTransactionTracker.cs b/print-service/src/Stickerlandia.PrintService.Core/Observability/DatadogTransactionTracker.cs new file mode 100644 index 00000000..4eb26d49 --- /dev/null +++ b/print-service/src/Stickerlandia.PrintService.Core/Observability/DatadogTransactionTracker.cs @@ -0,0 +1,109 @@ +// 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 2026 Datadog, Inc. + +using System.Diagnostics; +using System.Globalization; +using System.IO.Compression; +using System.Net.Http.Headers; +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; + +namespace Stickerlandia.PrintService.Core.Observability; + +public class DatadogTransactionTracker : IDatadogTransactionTracker +{ + private readonly string _service; + private readonly string _environment; + private readonly string _ddApiKey; + private readonly Uri _pipelineStatsEndpoint; + + private readonly IHttpClientFactory _httpClientFactory; + private readonly ILogger _logger; + + public DatadogTransactionTracker(IHttpClientFactory httpClientFactory, + IConfiguration configuration, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(configuration); + + _httpClientFactory = httpClientFactory; + _logger = logger; + _service = configuration["DD_SERVICE"] ?? "print-service"; + _environment = configuration["DD_ENV"] ?? "local"; + _ddApiKey = configuration["DD_API_KEY"] ?? ""; + _pipelineStatsEndpoint = new Uri($"https://trace.agent.{configuration["DD_SITE"] ?? "datadoghq.com"}/api/v0.1/pipeline_stats"); + } + + public async Task TrackTransactionAsync(string transactionId, string checkpoint) + { + if (string.IsNullOrEmpty(_ddApiKey)) + { + Log.TransactionTrackingSkipped(_logger); + return; + } + + try + { + if (Activity.Current is not null) + { + Activity.Current.SetTag("dsm.transaction_id", transactionId); + Activity.Current.SetTag("dsm.transaction.checkpoint", checkpoint); + } + + var timestampNanos = (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L) + .ToString(CultureInfo.InvariantCulture); + + var payload = new + { + transactions = new[] + { + new + { + transaction_id = transactionId, + checkpoint, + timestamp_nanos = timestampNanos + } + }, + service = _service, + environment = _environment + }; + + var json = JsonSerializer.Serialize(payload); + var compressed = Gzip(Encoding.UTF8.GetBytes(json)); + + using var client = _httpClientFactory.CreateClient(); + using var request = new HttpRequestMessage(HttpMethod.Post, _pipelineStatsEndpoint); + request.Content = new ByteArrayContent(compressed); + request.Content.Headers.ContentType = new MediaTypeHeaderValue("application/json"); + request.Content.Headers.ContentEncoding.Add("gzip"); + request.Content.Headers.Add("DD-API-KEY", _ddApiKey); + + var response = await client.SendAsync(request); + response.EnsureSuccessStatusCode(); + + #pragma warning disable + _logger.LogInformation("Successfully tracked transaction {TransactionId} at checkpoint {Checkpoint} with status code {StatusCode}", transactionId, checkpoint, response.StatusCode); + #pragma warning enable + } +#pragma warning disable CA1031 + catch (Exception ex) +#pragma warning restore CA1031 + { + Log.TransactionTrackingFailed(_logger, transactionId, checkpoint, ex); + } + } + + private static byte[] Gzip(byte[] data) + { + using var output = new MemoryStream(); + using (var gzip = new GZipStream(output, CompressionMode.Compress)) + { + gzip.Write(data); + } + + return output.ToArray(); + } +} diff --git a/print-service/src/Stickerlandia.PrintService.Core/Observability/IDatadogTransactionTracker.cs b/print-service/src/Stickerlandia.PrintService.Core/Observability/IDatadogTransactionTracker.cs new file mode 100644 index 00000000..d4824455 --- /dev/null +++ b/print-service/src/Stickerlandia.PrintService.Core/Observability/IDatadogTransactionTracker.cs @@ -0,0 +1,10 @@ +// 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 2026 Datadog, Inc. + +namespace Stickerlandia.PrintService.Core.Observability; + +public interface IDatadogTransactionTracker +{ + Task TrackTransactionAsync(string transactionId, string checkpoint); +} diff --git a/print-service/src/Stickerlandia.PrintService.Core/Observability/LogMessages.cs b/print-service/src/Stickerlandia.PrintService.Core/Observability/LogMessages.cs index 1a389ea3..061a86c1 100644 --- 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,13 +87,26 @@ 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); [LoggerMessage( - EventId = 5, + EventId = 11, + Level = LogLevel.Warning, + Message = "DD_API_KEY is not configured; skipping transaction tracking.")] + public static partial void TransactionTrackingSkipped(ILogger logger); + + [LoggerMessage( + EventId = 12, + Level = LogLevel.Warning, + Message = "Failed to track Datadog transaction {TransactionId} at checkpoint {Checkpoint}")] + public static partial void TransactionTrackingFailed( + ILogger logger, string transactionId, string checkpoint, Exception exception); + + [LoggerMessage( + EventId = 13, Level = LogLevel.Trace, Message = "Stopping message processor for transport: {MessageTransport}")] public static partial void StoppingMessageProcessor( diff --git a/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/AcknowledgePrintJobCommandHandler.cs b/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/AcknowledgePrintJobCommandHandler.cs index 33c532e0..224b3aac 100644 --- a/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/AcknowledgePrintJobCommandHandler.cs +++ b/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/AcknowledgePrintJobCommandHandler.cs @@ -15,7 +15,8 @@ public class AcknowledgePrintJobCommandHandler( IPrintJobRepository printJobRepository, IPrinterRepository printerRepository, IOutbox outbox, - PrintJobInstrumentation instrumentation) + PrintJobInstrumentation instrumentation, + IDatadogTransactionTracker transactionTracker) : ICommandHandler { public async Task Handle(AcknowledgePrintJobCommand command, CancellationToken cancellationToken = default) @@ -62,6 +63,7 @@ public async Task Handle(AcknowledgePrintJobCommand if (command.Success) { printJob.Complete(); + await transactionTracker.TrackTransactionAsync(printJob.Id.Value, "print-job-acknowledged"); await outbox.StoreEventFor(new PrintJobCompletedEvent(printJob)); // Record completion metrics diff --git a/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/PrintJobCompletedEvent.cs b/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/PrintJobCompletedEvent.cs index ddf08419..7aeee6db 100644 --- a/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/PrintJobCompletedEvent.cs +++ b/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/PrintJobCompletedEvent.cs @@ -15,8 +15,10 @@ public PrintJobCompletedEvent(PrintJob printJob) { ArgumentNullException.ThrowIfNull(printJob); + UserId = printJob.UserId; PrintJobId = printJob.Id.Value; PrinterId = printJob.PrinterId.Value; + StickerId = printJob.StickerId; CompletedAt = printJob.CompletedAt ?? DateTimeOffset.UtcNow; } @@ -34,6 +36,12 @@ public PrintJobCompletedEvent(PrintJob printJob) [JsonPropertyName("printerId")] public string PrinterId { get; set; } = string.Empty; + [JsonPropertyName("userId")] + public string UserId { get; set; } = string.Empty; + + [JsonPropertyName("stickerId")] + public string StickerId { get; set; } = string.Empty; + [JsonPropertyName("completedAt")] public DateTimeOffset CompletedAt { get; set; } } diff --git a/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/SubmitPrintJobCommandHandler.cs b/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/SubmitPrintJobCommandHandler.cs index f09e31dc..e796183c 100644 --- a/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/SubmitPrintJobCommandHandler.cs +++ b/print-service/src/Stickerlandia.PrintService.Core/PrintJobs/SubmitPrintJobCommandHandler.cs @@ -44,7 +44,9 @@ public async Task Handle(SubmitPrintJobCommand command, command.StickerId, command.StickerUrl); - // Inject DSM context at submission time so it's persisted with the job + // Inject DSM context at submission time so it's persisted with the job. + // An explicit producer span is created to guarantee InjectIncludingDsm always fires, + // regardless of whether an outer Datadog scope happens to be active at this call site. if (activity != null) { using var context = Tracer.Instance.StartActive("print_queue"); @@ -55,7 +57,7 @@ public async Task Handle(SubmitPrintJobCommand command, injector.InjectIncludingDsm(printJob, (_, key, value) => { printJob.AddHeader(key, value); - }, context.Span.Context, "print", "print_queue"); + }, context.Span.Context, "queue", "print_queue"); } await printJobRepository.AddAsync(printJob); diff --git a/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/EventBridgeEventPublisherTests.cs b/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/EventBridgeEventPublisherTests.cs index 665fd9a4..7725f32c 100644 --- a/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/EventBridgeEventPublisherTests.cs +++ b/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/EventBridgeEventPublisherTests.cs @@ -8,6 +8,7 @@ using Microsoft.Extensions.Options; using Stickerlandia.PrintService.AWS; using Stickerlandia.PrintService.Core; +using Stickerlandia.PrintService.Core.Observability; using Stickerlandia.PrintService.Core.PrintJobs; using Stickerlandia.PrintService.Core.RegisterPrinter; @@ -24,8 +25,9 @@ public EventBridgeEventPublisherTests() var logger = A.Fake>(); var options = A.Fake>(); A.CallTo(() => options.Value).Returns(new AwsConfiguration { EventBusName = "test-bus" }); + var transactionTracker = A.Fake(); - _publisher = new EventBridgeEventPublisher(logger, _eventBridgeClient, options); + _publisher = new EventBridgeEventPublisher(logger, _eventBridgeClient, options, transactionTracker); } [Fact] diff --git a/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/OutboxFunctionsTests.cs b/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/OutboxFunctionsTests.cs index 8b0efca2..1641a937 100644 --- a/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/OutboxFunctionsTests.cs +++ b/print-service/tests/Stickerlandia.PrintService.UnitTest/AwsTests/OutboxFunctionsTests.cs @@ -47,7 +47,8 @@ public OutboxFunctionsTests() var eventPublisher = new EventBridgeEventPublisher( A.Fake>(), _eventBridgeClient, - _awsConfiguration); + _awsConfiguration, + A.Fake()); _sut = new OutboxFunctions(_logger, outboxProcessor, eventPublisher); } diff --git a/print-service/tests/Stickerlandia.PrintService.UnitTest/PrintJobTests/AcknowledgePrintJobCommandHandlerTests.cs b/print-service/tests/Stickerlandia.PrintService.UnitTest/PrintJobTests/AcknowledgePrintJobCommandHandlerTests.cs index 11a2ce41..86f29608 100644 --- a/print-service/tests/Stickerlandia.PrintService.UnitTest/PrintJobTests/AcknowledgePrintJobCommandHandlerTests.cs +++ b/print-service/tests/Stickerlandia.PrintService.UnitTest/PrintJobTests/AcknowledgePrintJobCommandHandlerTests.cs @@ -18,6 +18,7 @@ public class AcknowledgePrintJobCommandHandlerTests : IDisposable private readonly IPrintJobRepository _printJobRepository; private readonly IPrinterRepository _printerRepository; private readonly IOutbox _outbox; + private readonly IDatadogTransactionTracker _transactionTracker; private readonly PrintJobInstrumentation _instrumentation; private readonly AcknowledgePrintJobCommandHandler _handler; @@ -26,8 +27,9 @@ public AcknowledgePrintJobCommandHandlerTests() _printJobRepository = A.Fake(); _printerRepository = A.Fake(); _outbox = A.Fake(); + _transactionTracker = A.Fake(); _instrumentation = new PrintJobInstrumentation(); - _handler = new AcknowledgePrintJobCommandHandler(_printJobRepository, _printerRepository, _outbox, _instrumentation); + _handler = new AcknowledgePrintJobCommandHandler(_printJobRepository, _printerRepository, _outbox, _instrumentation, _transactionTracker); } public void Dispose() diff --git a/user-management/infra/aws/lib/api.ts b/user-management/infra/aws/lib/api.ts index 3291af07..a088cab0 100644 --- a/user-management/infra/aws/lib/api.ts +++ b/user-management/infra/aws/lib/api.ts @@ -33,6 +33,8 @@ export class ApiProps { export class Api extends Construct { stickerClaimedQueue: Queue; stickerClaimedDLQ: Queue; + stickerPrintedQueue: Queue; + stickerPrintedDLQ: Queue; userRegisteredTopic: Topic; constructor(scope: Construct, id: string, props: ApiProps) { super(scope, id); @@ -44,7 +46,10 @@ export class Api extends Construct { queueName: `${props.sharedProps.serviceName}-${props.sharedProps.environment}-sticker-claimed-dlq`, }); - //TODO: Add EventBridge rule mapping to subscribe to sticker claimed events published to the shared EventBus. + this.stickerPrintedDLQ = new Queue(this, "StickerPrintedDLQ", { + queueName: `${props.sharedProps.serviceName}-${props.sharedProps.environment}-sticker-printed-dlq`, + }); + this.stickerClaimedQueue = new Queue(this, "StickerClaimedQueue", { queueName: `${props.sharedProps.serviceName}-${props.sharedProps.environment}-sticker-claimed`, deadLetterQueue: { @@ -52,6 +57,13 @@ export class Api extends Construct { maxReceiveCount: 5, // Messages will be sent to DLQ after 5 failed attempts }, }); + this.stickerPrintedQueue = new Queue(this, "StickerPrintedQueue", { + queueName: `${props.sharedProps.serviceName}-${props.sharedProps.environment}-sticker-printed`, + deadLetterQueue: { + queue: this.stickerPrintedDLQ, + maxReceiveCount: 5, // Messages will be sent to DLQ after 5 failed attempts + }, + }); const webService = new WebService(this, "UserServiceWebService", { sharedProps: props.sharedProps, @@ -121,6 +133,10 @@ export class Api extends Construct { this.stickerClaimedDLQ.grantSendMessages(webService.taskRole); this.stickerClaimedQueue.grantConsumeMessages(webService.taskRole); this.stickerClaimedDLQ.grantConsumeMessages(webService.taskRole); + this.stickerPrintedQueue.grantSendMessages(webService.taskRole); + this.stickerPrintedDLQ.grantSendMessages(webService.taskRole); + this.stickerPrintedQueue.grantConsumeMessages(webService.taskRole); + this.stickerPrintedDLQ.grantConsumeMessages(webService.taskRole); // Grant execution role permission to read the database connection string secret props.serviceProps.databaseCredentials.grantRead(webService.executionRole); diff --git a/user-management/infra/aws/lib/background-workers.ts b/user-management/infra/aws/lib/background-workers.ts index 6c4537ed..0a4a7585 100644 --- a/user-management/infra/aws/lib/background-workers.ts +++ b/user-management/infra/aws/lib/background-workers.ts @@ -41,7 +41,9 @@ export interface BackgroundWorkersProps { serviceProps: ServiceProps; sharedEventBus: IEventBus; stickerClaimedQueue: IQueue; + stickerPrintedQueue: IQueue; stickerClaimedDLQ: IQueue; + stickerPrintedDLQ: IQueue; userRegisteredTopic: ITopic; useLambda: boolean; } @@ -52,13 +54,14 @@ export class BackgroundWorkers extends Construct { if (props.useLambda) { // Get connection string value from CustomResource output, resolved at deploy time - const connectionString = props.serviceProps.databaseCredentials.getConnectionStringForLambda(); + const connectionString = + props.serviceProps.databaseCredentials.getConnectionStringForLambda(); // Reference the VPC link security group for Lambda functions that need database access const lambdaSecurityGroup = SecurityGroup.fromSecurityGroupId( this, "LambdaSecurityGroup", - props.vpcLinkSecurityGroupId + props.vpcLinkSecurityGroupId, ); // Lambda functions need to be in private subnets to access RDS @@ -80,6 +83,8 @@ export class BackgroundWorkers extends Construct { Aws__UserRegisteredTopicArn: props.userRegisteredTopic.topicArn, Aws__StickerClaimedQueueUrl: props.stickerClaimedQueue.queueUrl, Aws__StickerClaimedDLQUrl: props.stickerClaimedDLQ.queueUrl, + Aws__StickerPrintedQueueUrl: props.stickerPrintedQueue.queueUrl, + Aws__StickerPrintedDLQUrl: props.stickerPrintedDLQ.queueUrl, DRIVING: "ASPNET", DRIVEN: "AWS", DISABLE_SSL: "true", @@ -103,14 +108,14 @@ export class BackgroundWorkers extends Construct { vpc: props.vpc, vpcSubnets: vpcSubnets, securityGroups: [lambdaSecurityGroup], - } + }, ); stickerClaimedWorker.function.addEventSource( new SqsEventSource(props.stickerClaimedQueue, { batchSize: 10, reportBatchItemFailures: true, - }) + }), ); // Add dependencies for Lambda functions to ensure SSM parameters exist before deployment @@ -130,6 +135,50 @@ export class BackgroundWorkers extends Construct { }); rule.addTarget(new SqsQueue(props.stickerClaimedQueue)); + const stickerPrintedWorker = new InstrumentedLambdaFunction( + this, + "StickerPrintedWorkerFunction", + { + sharedProps: props.sharedProps, + handler: + "Stickerlandia.UserManagement.Lambda::Stickerlandia.UserManagement.Lambda.StickerPrintedSqsHandler_Handler_Generated::Handler", + buildDef: "../../src/Stickerlandia.UserManagement.Lambda/", + functionName: "sticker-printed-worker", + environment: environmentVariables, + memorySize: 1024, + timeout: Duration.seconds(25), + logLevel: props.sharedProps.environment === "prod" ? "WARN" : "INFO", + onFailure: new SqsDestination(props.stickerPrintedDLQ), + vpc: props.vpc, + vpcSubnets: vpcSubnets, + securityGroups: [lambdaSecurityGroup], + }, + ); + + stickerPrintedWorker.function.addEventSource( + new SqsEventSource(props.stickerPrintedQueue, { + batchSize: 10, + reportBatchItemFailures: true, + }), + ); + + // Add dependencies for Lambda functions to ensure SSM parameters exist before deployment + if (props.serviceProps.serviceDependencies) { + for (const dependency of props.serviceProps.serviceDependencies) { + stickerPrintedWorker.function.node.addDependency(dependency); + } + } + + const stickerPrintedRule = new Rule(this, "StickerPrintedEventRule", { + eventBus: props.sharedEventBus, + ruleName: `${props.sharedProps.serviceName}-${props.sharedProps.environment}-sticker-printed-rule`, + eventPattern: { + source: [`${props.sharedProps.environment}.print`], + detailType: ["printJobs.completed.v1"], + }, + }); + stickerPrintedRule.addTarget(new SqsQueue(props.stickerPrintedQueue)); + const outboxWorker = new InstrumentedLambdaFunction( this, "OutboxWorkerFunction", @@ -147,7 +196,7 @@ export class BackgroundWorkers extends Construct { vpc: props.vpc, vpcSubnets: vpcSubnets, securityGroups: [lambdaSecurityGroup], - } + }, ); props.sharedEventBus.grantPutEventsTo(outboxWorker.function); @@ -170,7 +219,7 @@ export class BackgroundWorkers extends Construct { event: RuleTargetInput.fromObject({ run: true, }), - }) + }), ); } else { const workerService = new WorkerService( @@ -223,6 +272,9 @@ export class BackgroundWorkers extends Construct { serviceDependencies: props.serviceProps.serviceDependencies, }, ); + + props.stickerClaimedQueue.grantConsumeMessages(workerService.taskRole); + props.stickerPrintedQueue.grantConsumeMessages(workerService.taskRole); } } } diff --git a/user-management/infra/aws/lib/user-service-stack.ts b/user-management/infra/aws/lib/user-service-stack.ts index 67292af6..660477fd 100644 --- a/user-management/infra/aws/lib/user-service-stack.ts +++ b/user-management/infra/aws/lib/user-service-stack.ts @@ -152,6 +152,8 @@ export class UserServiceStack extends cdk.Stack { serviceDiscoveryNamespace: sharedResources.serviceDiscoveryNamespace, stickerClaimedQueue: api.stickerClaimedQueue, stickerClaimedDLQ: api.stickerClaimedDLQ, + stickerPrintedQueue: api.stickerPrintedQueue, + stickerPrintedDLQ: api.stickerPrintedDLQ, userRegisteredTopic: api.userRegisteredTopic, deployInPrivateSubnet: true, }); diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/AwsConfiguration.cs b/user-management/src/Stickerlandia.UserManagement.AWS/AwsConfiguration.cs index ad23a0c6..98f4b740 100644 --- a/user-management/src/Stickerlandia.UserManagement.AWS/AwsConfiguration.cs +++ b/user-management/src/Stickerlandia.UserManagement.AWS/AwsConfiguration.cs @@ -22,4 +22,8 @@ public class AwsConfiguration public string StickerClaimedQueueUrl { get; set; } = string.Empty; public string StickerClaimedDLQUrl { get; set; } = string.Empty; + + public string StickerPrintedQueueUrl { get; set; } = string.Empty; + + public string StickerPrintedDLQUrl { get; set; } = string.Empty; } \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/EventBridgeEventPublisher.cs b/user-management/src/Stickerlandia.UserManagement.AWS/EventBridgeEventPublisher.cs index 1d00a743..fd9d4cf9 100644 --- a/user-management/src/Stickerlandia.UserManagement.AWS/EventBridgeEventPublisher.cs +++ b/user-management/src/Stickerlandia.UserManagement.AWS/EventBridgeEventPublisher.cs @@ -9,6 +9,7 @@ // Copyright 2025 Datadog, Inc. using System.Globalization; +using System.Text.Json.Nodes; using Amazon.EventBridge; using Amazon.EventBridge.Model; using Amazon.SimpleNotificationService; @@ -73,6 +74,14 @@ private async Task Publish(CloudEvent cloudEvent) var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); var jsonString = System.Text.Encoding.UTF8.GetString(data.Span); + if (Tracer.Instance.ActiveScope is not null) + { + var jsonNode = JsonNode.Parse(jsonString)!; + new SpanContextInjector().InjectIncludingDsm( + jsonNode, SetDsmHeader, Tracer.Instance.ActiveScope.Span.Context, "eventbridge", cloudEvent.Type!); + jsonString = jsonNode.ToJsonString(); + } + await client.PutEventsAsync(new PutEventsRequest() { Entries = new List(1) @@ -98,4 +107,10 @@ await client.PutEventsAsync(new PutEventsRequest() processScope?.Close(); } } + + private static void SetDsmHeader(JsonNode jsonNode, string key, string value) + { + if (jsonNode["_datadog"] == null) jsonNode["_datadog"] = new JsonObject(); + jsonNode["_datadog"]![key] = value; + } } \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/ServiceExtensions.cs b/user-management/src/Stickerlandia.UserManagement.AWS/ServiceExtensions.cs index 55c5e1d1..d38fd509 100644 --- a/user-management/src/Stickerlandia.UserManagement.AWS/ServiceExtensions.cs +++ b/user-management/src/Stickerlandia.UserManagement.AWS/ServiceExtensions.cs @@ -30,7 +30,8 @@ public static IServiceCollection AddAwsAdapters(this IServiceCollection services services.Configure( configuration.GetSection("Aws")); - services.AddSingleton(); + services.AddKeyedSingleton("stickerClaimed"); + services.AddKeyedSingleton("stickerPrinted"); services.AddSingleton(sp => new AmazonSQSClient()); services.AddSingleton(sp => new AmazonEventBridgeClient()); services.AddSingleton(sp => new AmazonSimpleNotificationServiceClient()); diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/SnsEventPublisher.cs b/user-management/src/Stickerlandia.UserManagement.AWS/SnsEventPublisher.cs index a63af899..af4797d8 100644 --- a/user-management/src/Stickerlandia.UserManagement.AWS/SnsEventPublisher.cs +++ b/user-management/src/Stickerlandia.UserManagement.AWS/SnsEventPublisher.cs @@ -71,7 +71,20 @@ private async Task Publish(CloudEvent cloudEvent) var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); var jsonString = System.Text.Encoding.UTF8.GetString(data.Span); - await client.PublishAsync(new PublishRequest(awsConfiguration.Value.UserRegisteredTopicArn, jsonString)); + var publishRequest = new PublishRequest(awsConfiguration.Value.UserRegisteredTopicArn, jsonString); + + if (Tracer.Instance.ActiveScope is not null) + { + publishRequest.MessageAttributes = new Dictionary(); + new SpanContextInjector().InjectIncludingDsm( + publishRequest.MessageAttributes, + (attrs, key, value) => attrs[key] = new MessageAttributeValue { DataType = "String", StringValue = value }, + Tracer.Instance.ActiveScope.Span.Context, + "sns", + awsConfiguration.Value.UserRegisteredTopicArn); + } + + await client.PublishAsync(publishRequest); } catch (Exception ex) { diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerClaimedWorker.cs b/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerClaimedWorker.cs index 8dcfca1d..86d05923 100644 --- a/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerClaimedWorker.cs +++ b/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerClaimedWorker.cs @@ -44,7 +44,19 @@ public SqsStickerClaimedWorker(ILogger logger, [SubscribeOperation(typeof(StickerClaimedEventV1))] private async Task ProcessMessageAsync(Message message) { - using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1"); + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + message.MessageAttributes, + static (attributes, key) => + { + if (attributes.TryGetValue(key, out var attr)) + return new[] { attr.StringValue }; + return Enumerable.Empty(); + }, + "sqs", + "users.stickerClaimed.v1"); + + using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1", + new SpanCreationSettings { Parent = extractedContext }); using var scope = _serviceScopeFactory.CreateScope(); var handler = scope.ServiceProvider.GetRequiredService(); @@ -84,8 +96,9 @@ public async Task PollAsync(CancellationToken stoppingToken) var request = new ReceiveMessageRequest { QueueUrl = _awsConfiguration.Value.StickerClaimedQueueUrl, - WaitTimeSeconds = 20, // Enable long polling with a 20-second wait time - MaxNumberOfMessages = 10 // Fetch up to 10 messages per call + WaitTimeSeconds = 20, + MaxNumberOfMessages = 10, + MessageAttributeNames = new List { "All" } }; var messages = await _sqsClient.ReceiveMessageAsync(request, stoppingToken); diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerPrintedWorker.cs new file mode 100644 index 00000000..c483bb48 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.AWS/SqsStickerPrintedWorker.cs @@ -0,0 +1,130 @@ +/* + * 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. + */ + +// 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 Datadog, Inc. + +using System.Text.Json; +using Amazon.Lambda.CloudWatchEvents; +using Amazon.SQS; +using Amazon.SQS.Model; +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; +using Datadog.Trace; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Saunter.Attributes; +using Stickerlandia.UserManagement.Core; +using Stickerlandia.UserManagement.Core.Observability; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; + +namespace Stickerlandia.UserManagement.AWS; + +[AsyncApi] +public class SqsStickerPrintedWorker : IMessagingWorker +{ + private readonly IServiceScopeFactory _serviceScopeFactory; + private readonly ILogger _logger; + private readonly AmazonSQSClient _sqsClient; + private readonly IOptions _awsConfiguration; + private readonly JsonSerializerOptions _jsonSerializerOptions = new() { PropertyNameCaseInsensitive = true }; + + public SqsStickerPrintedWorker(ILogger logger, + AmazonSQSClient sqsClient, IServiceScopeFactory serviceScopeFactory, + IOptions awsConfiguration) + { + _logger = logger; + _sqsClient = sqsClient; + _serviceScopeFactory = serviceScopeFactory; + _awsConfiguration = awsConfiguration; + } + + [Channel("printJobs.completed.v1")] + [SubscribeOperation(typeof(StickerPrintedEventV1))] + private async Task ProcessMessageAsync(Message message) + { + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + message.MessageAttributes, + static (attributes, key) => + { + if (attributes.TryGetValue(key, out var attr)) + return new[] { attr.StringValue }; + return Enumerable.Empty(); + }, + "sqs", + "printJobs.completed.v1"); + + using var processSpan = Tracer.Instance.StartActive($"process printJobs.completed.v1", + new SpanCreationSettings { Parent = extractedContext }); + + using var scope = _serviceScopeFactory.CreateScope(); + var handler = scope.ServiceProvider.GetRequiredService(); + + var envelope = JsonSerializer.Deserialize>(message.Body, _jsonSerializerOptions); + + if (envelope == null) + { + await _sqsClient.SendMessageAsync(_awsConfiguration.Value.StickerPrintedDLQUrl, message.Body); + await _sqsClient.DeleteMessageAsync(_awsConfiguration.Value.StickerPrintedQueueUrl, message.ReceiptHandle); + return; + } + + var detailBytes = JsonSerializer.SerializeToUtf8Bytes(envelope.Detail, _jsonSerializerOptions); + var formatter = new JsonEventFormatter(); + var cloudEvent = await formatter.DecodeStructuredModeMessageAsync( + new MemoryStream(detailBytes), null, new List()); + var evtData = (StickerPrintedEventV1?)cloudEvent.Data; + + if (evtData == null) + { + await _sqsClient.SendMessageAsync(_awsConfiguration.Value.StickerPrintedDLQUrl, message.Body); + await _sqsClient.DeleteMessageAsync(_awsConfiguration.Value.StickerPrintedQueueUrl, message.ReceiptHandle); + return; + } + + try + { + await handler.Handle(evtData); + } + catch (InvalidUserException ex) + { + Log.InvalidUser(_logger, ex); + + await _sqsClient.SendMessageAsync(_awsConfiguration.Value.StickerPrintedDLQUrl, message.Body); + } + + await _sqsClient.DeleteMessageAsync(_awsConfiguration.Value.StickerPrintedQueueUrl, message.ReceiptHandle); + } + + public Task StartAsync() + { + Log.StartingMessageProcessor(_logger, "sqs"); + return Task.CompletedTask; + } + + public async Task PollAsync(CancellationToken stoppingToken) + { + var request = new ReceiveMessageRequest + { + QueueUrl = _awsConfiguration.Value.StickerPrintedQueueUrl, + WaitTimeSeconds = 20, + MaxNumberOfMessages = 10, + MessageAttributeNames = new List { "All" } + }; + + var messages = await _sqsClient.ReceiveMessageAsync(request, stoppingToken); + foreach (var message in messages.Messages) await ProcessMessageAsync(message); + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + Log.StoppingMessageProcessor(_logger, "sqs"); + // No specific stop logic for SQS, as it is a pull-based system. + await Task.CompletedTask; + } +} \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.AWS/Stickerlandia.UserManagement.AWS.csproj b/user-management/src/Stickerlandia.UserManagement.AWS/Stickerlandia.UserManagement.AWS.csproj index 63902de6..eba45eae 100644 --- a/user-management/src/Stickerlandia.UserManagement.AWS/Stickerlandia.UserManagement.AWS.csproj +++ b/user-management/src/Stickerlandia.UserManagement.AWS/Stickerlandia.UserManagement.AWS.csproj @@ -8,6 +8,7 @@ + diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/IKafkaConsumerFactory.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/IKafkaConsumerFactory.cs new file mode 100644 index 00000000..1b3aa638 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/IKafkaConsumerFactory.cs @@ -0,0 +1,20 @@ +/* + * 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 Confluent.Kafka; + +namespace Stickerlandia.UserManagement.Agnostic; + +public interface IKafkaConsumerFactory +{ + IConsumer Create(ConsumerConfig config); +} + +public sealed class KafkaConsumerFactory : IKafkaConsumerFactory +{ + public IConsumer Create(ConsumerConfig config) + => new ConsumerBuilder(config).Build(); +} diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaEventPublisher.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaEventPublisher.cs index 5760332d..c3fcd324 100644 --- a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaEventPublisher.cs +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaEventPublisher.cs @@ -66,13 +66,28 @@ private async Task Publish(CloudEvent cloudEvent) var formatter = new JsonEventFormatter(); var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + var message = new Message + { + Key = cloudEvent.Id!, + Value = Encoding.UTF8.GetString(data.Span) + }; + + if (Tracer.Instance.ActiveScope is not null) + { + message.Headers = new Headers(); + new SpanContextInjector().InjectIncludingDsm( + message.Headers, + (headers, key, value) => headers.Add(key, Encoding.UTF8.GetBytes(value)), + Tracer.Instance.ActiveScope.Span.Context, + "kafka", + cloudEvent.Type!); + } + using var producer = new ProducerBuilder(config).Build(); try { - var deliveryReport = await producer.ProduceAsync(cloudEvent.Type, - new Message { Key = cloudEvent.Id!, Value = Encoding.UTF8.GetString(data.Span) }, - default); + var deliveryReport = await producer.ProduceAsync(cloudEvent.Type, message, default); if (deliveryReport.Status == PersistenceStatus.PossiblyPersisted) { // Handle potential timeout errors diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafakStickerClaimedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerClaimedWorker.cs similarity index 61% rename from user-management/src/Stickerlandia.UserManagement.Agnostic/KafakStickerClaimedWorker.cs rename to user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerClaimedWorker.cs index 738f671d..0b23488c 100644 --- a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafakStickerClaimedWorker.cs +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerClaimedWorker.cs @@ -4,10 +4,6 @@ * Copyright 2025-Present Datadog, Inc. */ -// 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 Datadog, Inc. - // Catching generic exceptions is not recommended, but in this case we want to catch all exceptions so that a failure in outbox processing does not crash the application. #pragma warning disable CA1031 @@ -24,22 +20,37 @@ namespace Stickerlandia.UserManagement.Agnostic; [AsyncApi] -public class KafakStickerClaimedWorker( - ILogger logger, +public class KafkaStickerClaimedWorker( + ILogger logger, IServiceScopeFactory serviceScopeFactory, ConsumerConfig consumerConfig, - ProducerConfig producerConfig) + ProducerConfig producerConfig, + IKafkaConsumerFactory consumerFactory) : IMessagingWorker { private const string topic = "users.stickerClaimed.v1"; private const string dlqTopic = "users.stickerClaimed.v1.dlq"; + private IConsumer? _consumer; + [Channel("users.stickerClaimed.v1")] [SubscribeOperation(typeof(StickerClaimedEventV1))] private async Task ProcessMessageAsync(StickerClaimedHandler handler, ConsumeResult consumeResult) { - using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1"); + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + consumeResult.Message.Headers, + static (headers, key) => + { + if (headers is null || !headers.TryGetLastBytes(key, out var bytes) || bytes is null) + return Enumerable.Empty(); + return new[] { System.Text.Encoding.UTF8.GetString(bytes) }; + }, + "kafka", + topic); + + using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1", + new SpanCreationSettings { Parent = extractedContext }); Log.ReceivedMessage(logger, "kafka"); @@ -82,66 +93,50 @@ public Task StartAsync() { Log.StartingMessageProcessor(logger, "kafka"); + _consumer = consumerFactory.Create(consumerConfig); + _consumer.Subscribe(topic); + return Task.CompletedTask; } public async Task PollAsync(CancellationToken stoppingToken) { - // Check for cancellation first if (stoppingToken.IsCancellationRequested) return; - using var scope = serviceScopeFactory.CreateScope(); - var handler = scope.ServiceProvider.GetRequiredService(); - try { - using var consumer = new ConsumerBuilder(consumerConfig) - .SetErrorHandler((_, e) => Log.MessageProcessingException(logger, e.Reason, null)) - .Build(); - - consumer.Subscribe(topic); + var consumeResult = _consumer!.Consume(TimeSpan.FromSeconds(2)); - try + if (consumeResult != null) { - var consumeResult = consumer.Consume(TimeSpan.FromSeconds(2)); + using var scope = serviceScopeFactory.CreateScope(); + var handler = scope.ServiceProvider.GetRequiredService(); - if (consumeResult != null) - { - var processResult = await ProcessMessageAsync(handler, consumeResult); + var processResult = await ProcessMessageAsync(handler, consumeResult); - if (processResult) consumer.Commit(consumeResult); - } - } - catch (ConsumeException ex) - { - Log.MessageProcessingException(logger, "Error consuming message", ex); - } - catch (OperationCanceledException) - { - // This is expected when the token is canceled - Log.TokenCancelled(logger); - } - catch (Exception ex) - { - Log.MessageProcessingException(logger, "Error consuming message", ex); - } - finally - { - // Ensure consumer is closed even if an exception occurs - consumer.Close(); + if (processResult) _consumer.Commit(consumeResult); } } + catch (ConsumeException ex) + { + Log.MessageProcessingException(logger, "Error consuming message", ex); + } + catch (OperationCanceledException) + { + Log.TokenCancelled(logger); + } catch (Exception ex) { - Log.FailureStartingWorker(logger, "kafka", ex); - throw; + Log.MessageProcessingException(logger, "Error consuming message", ex); } } public Task StopAsync(CancellationToken cancellationToken) { - // NOOP + _consumer?.Close(); + _consumer?.Dispose(); + _consumer = null; return Task.CompletedTask; } -} \ No newline at end of file +} diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerPrintedWorker.cs new file mode 100644 index 00000000..e43a5163 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/KafkaStickerPrintedWorker.cs @@ -0,0 +1,151 @@ +/* + * 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. + */ + +// Catching generic exceptions is not recommended, but in this case we want to catch all exceptions so that a failure in outbox processing does not crash the application. +#pragma warning disable CA1031 + +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; +using Confluent.Kafka; +using Datadog.Trace; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Saunter.Attributes; +using Stickerlandia.UserManagement.Core.Observability; +using Stickerlandia.UserManagement.Core; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; + +namespace Stickerlandia.UserManagement.Agnostic; + +[AsyncApi] +public class KafkaStickerPrintedWorker( + ILogger logger, + IServiceScopeFactory serviceScopeFactory, + ConsumerConfig consumerConfig, + ProducerConfig producerConfig, + IKafkaConsumerFactory consumerFactory) + : IMessagingWorker +{ + private const string topic = "printJobs.completed.v1"; + private const string dlqTopic = "printJobs.completed.v1.dlq"; + + private IConsumer? _consumer; + private IProducer? _dlqProducer; + + [Channel("printJobs.completed.v1")] + [SubscribeOperation(typeof(StickerPrintedEventV1))] + private async Task ProcessMessageAsync(StickerPrintedHandler handler, + ConsumeResult consumeResult) + { + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + consumeResult.Message.Headers, + static (headers, key) => + { + if (headers is null || !headers.TryGetLastBytes(key, out var bytes) || bytes is null) + return Enumerable.Empty(); + return new[] { System.Text.Encoding.UTF8.GetString(bytes) }; + }, + "kafka", + topic); + + using var processSpan = Tracer.Instance.StartActive($"process printJobs.completed.v1", + new SpanCreationSettings { Parent = extractedContext }); + + Log.ReceivedMessage(logger, "kafka"); + + var detailBytes = System.Text.Encoding.UTF8.GetBytes(consumeResult.Message.Value); + var formatter = new JsonEventFormatter(); + var cloudEvent = await formatter.DecodeStructuredModeMessageAsync( + new MemoryStream(detailBytes), null, new List()); + var stickerPrinted = (StickerPrintedEventV1?)cloudEvent.Data; + + if (stickerPrinted == null) return false; + + try + { + await handler.Handle(stickerPrinted); + } + catch (InvalidUserException ex) + { + Log.InvalidUser(logger, ex); + + SendToDLQ(consumeResult, stickerPrinted); + } + + return true; + } + + private void SendToDLQ(ConsumeResult consumeResult, StickerPrintedEventV1 evtData) + { + _dlqProducer!.Produce(dlqTopic, + new Message { Key = evtData.UserId, Value = consumeResult.Message.Value }, + (deliveryReport) => + { + if (deliveryReport.Error.Code != ErrorCode.NoError) + Log.MessageDeliveryFailure(logger, deliveryReport.Error.Reason, null); + else + Log.MessageSentToDlq(logger, "", null); + }); + } + + public Task StartAsync() + { + Log.StartingMessageProcessor(logger, "kafka"); + + _consumer = consumerFactory.Create(consumerConfig); + _consumer.Subscribe(topic); + + _dlqProducer = new ProducerBuilder(producerConfig).Build(); + + return Task.CompletedTask; + } + + public async Task PollAsync(CancellationToken stoppingToken) + { + if (stoppingToken.IsCancellationRequested) + return; + + try + { + var consumeResult = _consumer!.Consume(TimeSpan.FromSeconds(2)); + + if (consumeResult != null) + { + using var scope = serviceScopeFactory.CreateScope(); + var handler = scope.ServiceProvider.GetRequiredService(); + + var processResult = await ProcessMessageAsync(handler, consumeResult); + + if (processResult) _consumer.Commit(consumeResult); + } + } + catch (ConsumeException ex) + { + Log.MessageProcessingException(logger, "Error consuming message", ex); + } + catch (OperationCanceledException) + { + Log.TokenCancelled(logger); + } + catch (Exception ex) + { + Log.MessageProcessingException(logger, "Error consuming message", ex); + } + } + + 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/Migrations/20260225075545_AddPrintCount.Designer.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/20260225075545_AddPrintCount.Designer.cs new file mode 100644 index 00000000..cac5f414 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/20260225075545_AddPrintCount.Designer.cs @@ -0,0 +1,597 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; +using Stickerlandia.UserManagement.Agnostic; + +#nullable disable + +namespace Stickerlandia.UserManagement.Agnostic.Migrations +{ + [DbContext(typeof(UserManagementDbContext))] + [Migration("20260225075545_AddPrintCount")] + partial class AddPrintCount + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.1") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityRole", b => + { + b.Property("Id") + .HasColumnType("text"); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .HasColumnType("text"); + + b.Property("Name") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NormalizedName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.HasKey("Id"); + + b.HasIndex("NormalizedName") + .IsUnique() + .HasDatabaseName("RoleNameIndex"); + + b.ToTable("AspNetRoles", (string)null); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityRoleClaim", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("ClaimType") + .HasColumnType("text"); + + b.Property("ClaimValue") + .HasColumnType("text"); + + b.Property("RoleId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("RoleId"); + + b.ToTable("AspNetRoleClaims", (string)null); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserClaim", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("ClaimType") + .HasColumnType("text"); + + b.Property("ClaimValue") + .HasColumnType("text"); + + b.Property("UserId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("UserId"); + + b.ToTable("AspNetUserClaims", (string)null); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserLogin", b => + { + b.Property("LoginProvider") + .HasColumnType("text"); + + b.Property("ProviderKey") + .HasColumnType("text"); + + b.Property("ProviderDisplayName") + .HasColumnType("text"); + + b.Property("UserId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("LoginProvider", "ProviderKey"); + + b.HasIndex("UserId"); + + b.ToTable("AspNetUserLogins", (string)null); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserRole", b => + { + b.Property("UserId") + .HasColumnType("text"); + + b.Property("RoleId") + .HasColumnType("text"); + + b.HasKey("UserId", "RoleId"); + + b.HasIndex("RoleId"); + + b.ToTable("AspNetUserRoles", (string)null); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserToken", b => + { + b.Property("UserId") + .HasColumnType("text"); + + b.Property("LoginProvider") + .HasColumnType("text"); + + b.Property("Name") + .HasColumnType("text"); + + b.Property("Value") + .HasColumnType("text"); + + b.HasKey("UserId", "LoginProvider", "Name"); + + b.ToTable("AspNetUserTokens", (string)null); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreApplication", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("text"); + + b.Property("ApplicationType") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("ClientId") + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("ClientSecret") + .HasColumnType("text"); + + b.Property("ClientType") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("ConcurrencyToken") + .IsConcurrencyToken() + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("ConsentType") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("DisplayName") + .HasColumnType("text"); + + b.Property("DisplayNames") + .HasColumnType("text"); + + b.Property("JsonWebKeySet") + .HasColumnType("text"); + + b.Property("Permissions") + .HasColumnType("text"); + + b.Property("PostLogoutRedirectUris") + .HasColumnType("text"); + + b.Property("Properties") + .HasColumnType("text"); + + b.Property("RedirectUris") + .HasColumnType("text"); + + b.Property("Requirements") + .HasColumnType("text"); + + b.Property("Settings") + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("ClientId") + .IsUnique(); + + b.ToTable("OpenIddictApplications", (string)null); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreAuthorization", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("text"); + + b.Property("ApplicationId") + .HasColumnType("text"); + + b.Property("ConcurrencyToken") + .IsConcurrencyToken() + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("CreationDate") + .HasColumnType("timestamp with time zone"); + + b.Property("Properties") + .HasColumnType("text"); + + b.Property("Scopes") + .HasColumnType("text"); + + b.Property("Status") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("Subject") + .HasMaxLength(400) + .HasColumnType("character varying(400)"); + + b.Property("Type") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.HasKey("Id"); + + b.HasIndex("ApplicationId", "Status", "Subject", "Type"); + + b.ToTable("OpenIddictAuthorizations", (string)null); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreScope", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("text"); + + b.Property("ConcurrencyToken") + .IsConcurrencyToken() + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("Description") + .HasColumnType("text"); + + b.Property("Descriptions") + .HasColumnType("text"); + + b.Property("DisplayName") + .HasColumnType("text"); + + b.Property("DisplayNames") + .HasColumnType("text"); + + b.Property("Name") + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("Properties") + .HasColumnType("text"); + + b.Property("Resources") + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("Name") + .IsUnique(); + + b.ToTable("OpenIddictScopes", (string)null); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreToken", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("text"); + + b.Property("ApplicationId") + .HasColumnType("text"); + + b.Property("AuthorizationId") + .HasColumnType("text"); + + b.Property("ConcurrencyToken") + .IsConcurrencyToken() + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("CreationDate") + .HasColumnType("timestamp with time zone"); + + b.Property("ExpirationDate") + .HasColumnType("timestamp with time zone"); + + b.Property("Payload") + .HasColumnType("text"); + + b.Property("Properties") + .HasColumnType("text"); + + b.Property("RedemptionDate") + .HasColumnType("timestamp with time zone"); + + b.Property("ReferenceId") + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("Status") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("Subject") + .HasMaxLength(400) + .HasColumnType("character varying(400)"); + + b.Property("Type") + .HasMaxLength(150) + .HasColumnType("character varying(150)"); + + b.HasKey("Id"); + + b.HasIndex("AuthorizationId"); + + b.HasIndex("ReferenceId") + .IsUnique(); + + b.HasIndex("ApplicationId", "Status", "Subject", "Type"); + + b.ToTable("OpenIddictTokens", (string)null); + }); + + modelBuilder.Entity("Stickerlandia.UserManagement.Agnostic.PostgresOutboxItem", b => + { + b.Property("Id") + .HasColumnType("text") + .HasColumnName("id"); + + b.Property("EmailAddress") + .IsRequired() + .HasColumnType("text") + .HasColumnName("email_address"); + + b.Property("EventData") + .IsRequired() + .HasColumnType("text") + .HasColumnName("event_data"); + + b.Property("EventTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("event_time"); + + b.Property("EventType") + .IsRequired() + .HasColumnType("text") + .HasColumnName("event_type"); + + b.Property("Failed") + .HasColumnType("boolean") + .HasColumnName("failed"); + + b.Property("FailureReason") + .HasColumnType("text") + .HasColumnName("failure_reason"); + + b.Property("Processed") + .HasColumnType("boolean") + .HasColumnName("processed"); + + b.Property("TraceId") + .HasColumnType("text") + .HasColumnName("trace_id"); + + b.HasKey("Id"); + + b.ToTable("outbox_items", (string)null); + }); + + modelBuilder.Entity("Stickerlandia.UserManagement.Core.PostgresUserAccount", b => + { + b.Property("Id") + .HasColumnType("text") + .HasColumnName("id"); + + b.Property("AccessFailedCount") + .HasColumnType("integer"); + + b.Property("AccountTier") + .HasColumnType("integer") + .HasColumnName("account_tier"); + + b.Property("AccountType") + .HasColumnType("integer") + .HasColumnName("account_type"); + + b.Property("ClaimedStickerCount") + .HasColumnType("integer") + .HasColumnName("claimed_sticker_count"); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .HasColumnType("text"); + + b.Property("DateCreated") + .HasColumnType("timestamp with time zone") + .HasColumnName("date_created"); + + b.Property("Email") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("EmailConfirmed") + .HasColumnType("boolean"); + + b.Property("FirstName") + .IsRequired() + .HasColumnType("text") + .HasColumnName("first_name"); + + b.Property("LastName") + .IsRequired() + .HasColumnType("text") + .HasColumnName("last_name"); + + b.Property("LockoutEnabled") + .HasColumnType("boolean"); + + b.Property("LockoutEnd") + .HasColumnType("timestamp with time zone"); + + b.Property("NormalizedEmail") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NormalizedUserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("PasswordHash") + .HasColumnType("text"); + + b.Property("PhoneNumber") + .HasColumnType("text"); + + b.Property("PhoneNumberConfirmed") + .HasColumnType("boolean"); + + b.Property("PrintedStickerCount") + .HasColumnType("integer"); + + b.Property("SecurityStamp") + .HasColumnType("text"); + + b.Property("TwoFactorEnabled") + .HasColumnType("boolean"); + + b.Property("UserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.HasKey("Id"); + + b.HasIndex("NormalizedEmail") + .HasDatabaseName("EmailIndex"); + + b.HasIndex("NormalizedUserName") + .IsUnique() + .HasDatabaseName("UserNameIndex"); + + b.ToTable("users", (string)null); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityRoleClaim", b => + { + b.HasOne("Microsoft.AspNetCore.Identity.IdentityRole", null) + .WithMany() + .HasForeignKey("RoleId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserClaim", b => + { + b.HasOne("Stickerlandia.UserManagement.Core.PostgresUserAccount", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserLogin", b => + { + b.HasOne("Stickerlandia.UserManagement.Core.PostgresUserAccount", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserRole", b => + { + b.HasOne("Microsoft.AspNetCore.Identity.IdentityRole", null) + .WithMany() + .HasForeignKey("RoleId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("Stickerlandia.UserManagement.Core.PostgresUserAccount", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserToken", b => + { + b.HasOne("Stickerlandia.UserManagement.Core.PostgresUserAccount", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreAuthorization", b => + { + b.HasOne("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreApplication", "Application") + .WithMany("Authorizations") + .HasForeignKey("ApplicationId"); + + b.Navigation("Application"); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreToken", b => + { + b.HasOne("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreApplication", "Application") + .WithMany("Tokens") + .HasForeignKey("ApplicationId"); + + b.HasOne("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreAuthorization", "Authorization") + .WithMany("Tokens") + .HasForeignKey("AuthorizationId"); + + b.Navigation("Application"); + + b.Navigation("Authorization"); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreApplication", b => + { + b.Navigation("Authorizations"); + + b.Navigation("Tokens"); + }); + + modelBuilder.Entity("OpenIddict.EntityFrameworkCore.Models.OpenIddictEntityFrameworkCoreAuthorization", b => + { + b.Navigation("Tokens"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/20260225075545_AddPrintCount.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/20260225075545_AddPrintCount.cs new file mode 100644 index 00000000..d8bdf7ee --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/20260225075545_AddPrintCount.cs @@ -0,0 +1,30 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#pragma warning disable +#nullable disable + +namespace Stickerlandia.UserManagement.Agnostic.Migrations +{ + /// + public partial class AddPrintCount : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "PrintedStickerCount", + table: "users", + type: "integer", + nullable: false, + defaultValue: 0); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "PrintedStickerCount", + table: "users"); + } + } +} diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/UserManagementDbContextModelSnapshot.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/UserManagementDbContextModelSnapshot.cs index ad5fd690..a6b2efa8 100644 --- a/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/UserManagementDbContextModelSnapshot.cs +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/Migrations/UserManagementDbContextModelSnapshot.cs @@ -1,11 +1,4 @@ -/* - * 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. - */ - -#pragma warning disable - +// using System; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Infrastructure; @@ -484,6 +477,9 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("PhoneNumberConfirmed") .HasColumnType("boolean"); + b.Property("PrintedStickerCount") + .HasColumnType("integer"); + b.Property("SecurityStamp") .HasColumnType("text"); diff --git a/user-management/src/Stickerlandia.UserManagement.Agnostic/ServiceExtensions.cs b/user-management/src/Stickerlandia.UserManagement.Agnostic/ServiceExtensions.cs index 7341d958..a7a80a4b 100644 --- a/user-management/src/Stickerlandia.UserManagement.Agnostic/ServiceExtensions.cs +++ b/user-management/src/Stickerlandia.UserManagement.Agnostic/ServiceExtensions.cs @@ -88,10 +88,12 @@ public static IServiceCollection AddKafkaMessaging(this IServiceCollection servi services.AddSingleton(producerConfig); services.AddSingleton(consumerConfig); + services.AddSingleton(); // Register event publisher as singleton services.AddSingleton(); - services.AddSingleton(); + services.AddKeyedSingleton("stickerClaimed"); + services.AddKeyedSingleton("stickerPrinted"); return services; } diff --git a/user-management/src/Stickerlandia.UserManagement.Aspire/AppBuilderExtensions.cs b/user-management/src/Stickerlandia.UserManagement.Aspire/AppBuilderExtensions.cs index a4976363..4d785a6b 100644 --- a/user-management/src/Stickerlandia.UserManagement.Aspire/AppBuilderExtensions.cs +++ b/user-management/src/Stickerlandia.UserManagement.Aspire/AppBuilderExtensions.cs @@ -10,8 +10,11 @@ #pragma warning disable CA2012 // Accessing ValueTasks directly is ok in the Aspire project +using System.Text; using System.Text.Json; using Aspire.Hosting.AWS.Lambda; +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; using Aspire.Hosting.Azure; using Azure.Messaging.ServiceBus; using Confluent.Kafka; @@ -142,7 +145,9 @@ await adminClient.CreateTopicsAsync(new TopicSpecification[] { new() { Name = "users.stickerClaimed.v1", NumPartitions = 1, ReplicationFactor = 1 }, new() { Name = "users.stickerClaimed.v1.dlq", NumPartitions = 1, ReplicationFactor = 1 }, - new() { Name = "users.userRegistered.v1", NumPartitions = 1, ReplicationFactor = 1 } + new() { Name = "users.userRegistered.v1", NumPartitions = 1, ReplicationFactor = 1 }, + new() { Name = "printJobs.completed.v1", NumPartitions = 1, ReplicationFactor = 1 }, + new() { Name = "printJobs.completed.v1.dlq", NumPartitions = 1, ReplicationFactor = 1 }, }); } catch (CreateTopicsException e) @@ -177,13 +182,53 @@ public static IResourceBuilder WithKafkaTestCommands( var config = c.ServiceProvider.GetRequiredService(); using var producer = new ProducerBuilder(config).Build(); - await producer.ProduceAsync("users.stickerClaimed.v1", new Message + var cloudEvent = new CloudEvent(CloudEventsSpecVersion.V1_0) { - Key = "", Value = JsonSerializer.Serialize(new + Id = Guid.NewGuid().ToString(), + Source = new Uri("https://stickerlandia.com"), + Type = "users.stickerClaimed.v1", + Time = DateTime.UtcNow, + Data = JsonDocument.Parse(JsonSerializer.Serialize(new { accountId = TEST_COMMAND_ACCOUNT_ID, stickerId = TEST_COMMAND_STICKER_ID - }) + })).RootElement + }; + var formatter = new JsonEventFormatter(); + var encoded = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + + await producer.ProduceAsync("users.stickerClaimed.v1", new Message + { + Key = "", Value = Encoding.UTF8.GetString(encoded.Span) + }); + + producer.Flush(TimeSpan.FromSeconds(10)); + + return new ExecuteCommandResult { Success = true }; + }, new CommandOptions()); + + builder.WithCommand("SendStickerPrintedMessage", "Send Sticker Printed Message", async (c) => + { + var config = c.ServiceProvider.GetRequiredService(); + using var producer = new ProducerBuilder(config).Build(); + + var cloudEvent = new CloudEvent(CloudEventsSpecVersion.V1_0) + { + Id = Guid.NewGuid().ToString(), + Source = new Uri("https://stickerlandia.com"), + Type = "printJobs.completed.v1", + Time = DateTime.UtcNow, + Data = JsonDocument.Parse(JsonSerializer.Serialize(new + { + userId = TEST_COMMAND_ACCOUNT_ID, + })).RootElement + }; + var formatter = new JsonEventFormatter(); + var encoded = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + + await producer.ProduceAsync("printJobs.completed.v1", new Message + { + Key = "", Value = Encoding.UTF8.GetString(encoded.Span) }); producer.Flush(TimeSpan.FromSeconds(10)); @@ -208,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"); @@ -243,12 +291,27 @@ public static IResourceBuilder WithServiceBusTestC builder.WithCommand("SendSbMessage", "Send Service Bus message", async (c) => { var sbClient = c.ServiceProvider.GetRequiredService(); - await sbClient.CreateSender(builder.Resource.QueueName) - .SendMessageAsync(new ServiceBusMessage(JsonSerializer.Serialize(new + + var cloudEvent = new CloudEvent(CloudEventsSpecVersion.V1_0) + { + Id = Guid.NewGuid().ToString(), + Source = new Uri("https://stickerlandia.com"), + Type = builder.Resource.QueueName, + Time = DateTime.UtcNow, + Data = JsonDocument.Parse(JsonSerializer.Serialize(new { accountId = TEST_COMMAND_ACCOUNT_ID, stickerId = TEST_COMMAND_STICKER_ID - }))); + })).RootElement + }; + var formatter = new JsonEventFormatter(); + var encoded = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + + await sbClient.CreateSender(builder.Resource.QueueName) + .SendMessageAsync(new ServiceBusMessage(encoded) + { + ContentType = "application/cloudevents+json" + }); return new ExecuteCommandResult { Success = true }; }, new CommandOptions()); diff --git a/user-management/src/Stickerlandia.UserManagement.Aspire/Stickerlandia.UserManagement.Aspire.csproj b/user-management/src/Stickerlandia.UserManagement.Aspire/Stickerlandia.UserManagement.Aspire.csproj index ad2764d0..7a5a8aac 100644 --- a/user-management/src/Stickerlandia.UserManagement.Aspire/Stickerlandia.UserManagement.Aspire.csproj +++ b/user-management/src/Stickerlandia.UserManagement.Aspire/Stickerlandia.UserManagement.Aspire.csproj @@ -20,6 +20,7 @@ + all runtime; build; native; contentfiles; analyzers; buildtransitive diff --git a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusEventPublisher.cs b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusEventPublisher.cs index ec3f5a3e..06bb034c 100644 --- a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusEventPublisher.cs +++ b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusEventPublisher.cs @@ -73,6 +73,16 @@ private async Task Publish(CloudEvent cloudEvent) ContentType = "application/json" }; + if (Tracer.Instance.ActiveScope is not null) + { + new SpanContextInjector().InjectIncludingDsm( + serviceBusMessage.ApplicationProperties, + (properties, key, value) => properties[key] = value, + Tracer.Instance.ActiveScope.Span.Context, + "servicebus", + cloudEvent.Type!); + } + await sender.SendMessageAsync(serviceBusMessage); } catch (Exception ex) diff --git a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerClaimedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerClaimedWorker.cs index c9c54527..79bc4b67 100644 --- a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerClaimedWorker.cs +++ b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerClaimedWorker.cs @@ -47,7 +47,19 @@ public ServiceBusStickerClaimedWorker(ILogger lo [SubscribeOperation(typeof(StickerClaimedEventV1))] private async Task ProcessMessageAsync(ProcessMessageEventArgs args) { - using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1"); + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + args.Message.ApplicationProperties, + static (properties, key) => + { + if (properties.TryGetValue(key, out var value)) + return new[] { value.ToString()! }; + return Enumerable.Empty(); + }, + "servicebus", + "users.stickerClaimed.v1"); + + using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1", + new SpanCreationSettings { Parent = extractedContext }); using var scope = _serviceScopeFactory.CreateScope(); var handler = scope.ServiceProvider.GetRequiredService(); @@ -55,7 +67,18 @@ private async Task ProcessMessageAsync(ProcessMessageEventArgs args) var messageBody = args.Message.Body.ToString(); Log.ReceivedMessage(_logger, "ServiceBus"); - var evtData = JsonSerializer.Deserialize(messageBody); + StickerClaimedEventV1? evtData; + if (string.Equals(args.Message.ContentType, "application/cloudevents+json", StringComparison.OrdinalIgnoreCase)) + { + using var document = JsonDocument.Parse(messageBody); + evtData = document.RootElement.TryGetProperty("data", out var dataElement) + ? JsonSerializer.Deserialize(dataElement.GetRawText()) + : null; + } + else + { + evtData = JsonSerializer.Deserialize(messageBody); + } if (evtData == null) await args.DeadLetterMessageAsync(args.Message, "Message body cannot be deserialized"); diff --git a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerPrintedWorker.cs new file mode 100644 index 00000000..b4de00bd --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceBusStickerPrintedWorker.cs @@ -0,0 +1,128 @@ +/* + * 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. + */ + +// 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 Datadog, Inc. + +using System.Text.Json; +using Azure.Messaging.ServiceBus; +using Datadog.Trace; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Saunter.Attributes; +using Stickerlandia.UserManagement.Core; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; +using Log = Stickerlandia.UserManagement.Core.Observability.Log; + +namespace Stickerlandia.UserManagement.Azure; + +[AsyncApi] +public class ServiceBusStickerPrintedWorker : IMessagingWorker +{ + private readonly IServiceScopeFactory _serviceScopeFactory; + private readonly ILogger _logger; + private readonly ServiceBusProcessor _processor; + + public ServiceBusStickerPrintedWorker(ILogger logger, + ServiceBusClient serviceBusClient, IServiceScopeFactory serviceScopeFactory) + { + ArgumentNullException.ThrowIfNull(serviceBusClient, nameof(serviceBusClient)); + + _logger = logger; + _serviceScopeFactory = serviceScopeFactory; + + // Create the processor + _processor = serviceBusClient.CreateProcessor("printJobs.completed.v1"); + + // Set up handlers + _processor.ProcessMessageAsync += ProcessMessageAsync; + _processor.ProcessErrorAsync += ProcessErrorAsync; + } + + [Channel("printJobs.completed.v1")] + [SubscribeOperation(typeof(StickerPrintedEventV1))] + private async Task ProcessMessageAsync(ProcessMessageEventArgs args) + { + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + args.Message.ApplicationProperties, + static (properties, key) => + { + if (properties.TryGetValue(key, out var value)) + return new[] { value.ToString()! }; + return Enumerable.Empty(); + }, + "servicebus", + "printJobs.completed.v1"); + + using var processSpan = Tracer.Instance.StartActive($"process printJobs.completed.v1", + new SpanCreationSettings { Parent = extractedContext }); + + using var scope = _serviceScopeFactory.CreateScope(); + var handler = scope.ServiceProvider.GetRequiredService(); + + var messageBody = args.Message.Body.ToString(); + Log.ReceivedMessage(_logger, "ServiceBus"); + + 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(dataElement.GetRawText()) + : null; + } + else + { + stickerPrinted = JsonSerializer.Deserialize(messageBody); + } + + if (stickerPrinted == null) + { + await args.DeadLetterMessageAsync(args.Message, "Message body cannot be deserialized"); + return; + } + + try + { + await handler.Handle(stickerPrinted!); + } + catch (InvalidUserException ex) + { + Log.InvalidUser(_logger, ex); + await args.DeadLetterMessageAsync(args.Message, "Invalid account id"); + return; + } + + // Complete the message + await args.CompleteMessageAsync(args.Message, CancellationToken.None); + } + + private Task ProcessErrorAsync(ProcessErrorEventArgs args) + { + Log.MessageProcessingException(_logger, args.ErrorSource.ToString(), null); + return Task.CompletedTask; + } + + public Task StartAsync() + { + Log.StartingMessageProcessor(_logger, "ServiceBus"); + _processor.StartProcessingAsync(); + return Task.CompletedTask; + } + + public Task PollAsync(CancellationToken stoppingToken) + { + // This should be a no-op; + return Task.CompletedTask; + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + await _processor.StopProcessingAsync(cancellationToken); + await _processor.CloseAsync(cancellationToken); + } +} \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceExtensions.cs b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceExtensions.cs index e6cfafb5..54668c5c 100644 --- a/user-management/src/Stickerlandia.UserManagement.Azure/ServiceExtensions.cs +++ b/user-management/src/Stickerlandia.UserManagement.Azure/ServiceExtensions.cs @@ -24,7 +24,8 @@ public static IServiceCollection AddAzureAdapters(this IServiceCollection servic services.AddPostgresAuthServices(configuration, enableDefaultUi); - services.AddSingleton(); + services.AddKeyedSingleton("stickerClaimed"); + services.AddKeyedSingleton("stickerPrinted"); services.AddSingleton(sp => new ServiceBusClient(configuration["ConnectionStrings:messaging"])); diff --git a/user-management/src/Stickerlandia.UserManagement.Core/Observability/LogMessages.cs b/user-management/src/Stickerlandia.UserManagement.Core/Observability/LogMessages.cs index 8845fd23..ab614a75 100644 --- a/user-management/src/Stickerlandia.UserManagement.Core/Observability/LogMessages.cs +++ b/user-management/src/Stickerlandia.UserManagement.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); diff --git a/user-management/src/Stickerlandia.UserManagement.Core/PostgresUserAccount.cs b/user-management/src/Stickerlandia.UserManagement.Core/PostgresUserAccount.cs index 2ba185ae..2973c945 100644 --- a/user-management/src/Stickerlandia.UserManagement.Core/PostgresUserAccount.cs +++ b/user-management/src/Stickerlandia.UserManagement.Core/PostgresUserAccount.cs @@ -17,6 +17,7 @@ public class PostgresUserAccount : IdentityUser public string FirstName { get; set; } = string.Empty; public string LastName { get; set; } = string.Empty; public int ClaimedStickerCount { get; set; } + public int PrintedStickerCount { get; set; } public DateTime DateCreated { get; set; } public AccountTier AccountTier { get; set; } public AccountType AccountType { get; set; } @@ -55,4 +56,9 @@ public void StickerOrdered() { ClaimedStickerCount++; } + + public void StickerPrinted() + { + PrintedStickerCount++; + } } \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Core/ServiceExtensions.cs b/user-management/src/Stickerlandia.UserManagement.Core/ServiceExtensions.cs index 332098be..a2e4814a 100644 --- a/user-management/src/Stickerlandia.UserManagement.Core/ServiceExtensions.cs +++ b/user-management/src/Stickerlandia.UserManagement.Core/ServiceExtensions.cs @@ -13,6 +13,7 @@ using Stickerlandia.UserManagement.Core.Outbox; using Stickerlandia.UserManagement.Core.RegisterUser; using Stickerlandia.UserManagement.Core.StickerClaimedEvent; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; using Stickerlandia.UserManagement.Core.UpdateUserDetails; namespace Stickerlandia.UserManagement.Core; @@ -25,6 +26,7 @@ public static IServiceCollection AddStickerlandiaUserManagement(this IServiceCol services.AddTransient(); services.AddTransient(); services.AddTransient(); + services.AddTransient(); services.AddTransient(); diff --git a/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedEventV1.cs b/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedEventV1.cs new file mode 100644 index 00000000..0bad855e --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedEventV1.cs @@ -0,0 +1,19 @@ +/* + * 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. + */ + +// 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 Datadog, Inc. + +using System.Text.Json.Serialization; + +namespace Stickerlandia.UserManagement.Core.StickerPrintedEvent; + +public record StickerPrintedEventV1 +{ + [JsonPropertyName("userId")] + public string UserId {get;init;} = ""; +} \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedHandler.cs b/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedHandler.cs new file mode 100644 index 00000000..d407c5f0 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Core/StickerPrintedEvent/StickerPrintedHandler.cs @@ -0,0 +1,46 @@ +/* + * 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. + */ + +// 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 Datadog, Inc. + +using System.Diagnostics; +using Microsoft.AspNetCore.Identity; + +namespace Stickerlandia.UserManagement.Core.StickerPrintedEvent; + +public class StickerPrintedHandler(UserManager users) +{ + public async Task Handle(StickerPrintedEventV1 eventV1) + { + try + { + if (eventV1 == null || string.IsNullOrEmpty(eventV1.UserId)) + { + throw new ArgumentException("Invalid StickerPrintedEventV1"); + } + + var account = await users.FindByIdAsync(eventV1.UserId); + + if (account is null) + { + throw new InvalidUserException("No user found with the provided account ID."); + } + + account.StickerPrinted(); + + await users.UpdateAsync(account); + } + catch (Exception ex) + { + Activity.Current?.AddTag("stickerPrint.failed", true); + Activity.Current?.AddTag("error.message", ex.Message); + + throw; + } + } +} \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Core/UserAccount.cs b/user-management/src/Stickerlandia.UserManagement.Core/UserAccount.cs index ce641c8f..31141cab 100644 --- a/user-management/src/Stickerlandia.UserManagement.Core/UserAccount.cs +++ b/user-management/src/Stickerlandia.UserManagement.Core/UserAccount.cs @@ -60,7 +60,9 @@ public static UserAccount Register(string emailAddress, string password, string DateCreated = DateTime.UtcNow, AccountTier = AccountTier.Std, FirstName = firstName, - LastName = lastName + LastName = lastName, + ClaimedStickerCount = 0, + PrintedStickerCount = 0 }; userAccount._domainEvents.Add(new UserRegisteredEvent(userAccount)); @@ -74,6 +76,7 @@ public static UserAccount From( string firstName, string lastName, int claimedStickerCount, + int printedStickerCount, DateTime dateCreated, AccountTier accountTier, AccountType accountType) @@ -87,7 +90,8 @@ public static UserAccount From( AccountType = accountType, FirstName = firstName, LastName = lastName, - ClaimedStickerCount = claimedStickerCount + ClaimedStickerCount = claimedStickerCount, + PrintedStickerCount = printedStickerCount }; } @@ -112,9 +116,23 @@ public static UserAccount From( public IReadOnlyCollection DomainEvents => _domainEvents; public int ClaimedStickerCount { get; private set; } + + public int PrintedStickerCount { get; private set; } internal bool Changed { get; private set; } + internal void StickerPrinted() + { + PrintedStickerCount++; + Changed = true; + } + + internal void StickerClaimed() + { + ClaimedStickerCount++; + Changed = true; + } + internal static bool IsValidEmail(string email) { if (string.IsNullOrWhiteSpace(email)) diff --git a/user-management/src/Stickerlandia.UserManagement.Core/UserAccountDTO.cs b/user-management/src/Stickerlandia.UserManagement.Core/UserAccountDTO.cs index ae38a15c..585636a8 100644 --- a/user-management/src/Stickerlandia.UserManagement.Core/UserAccountDTO.cs +++ b/user-management/src/Stickerlandia.UserManagement.Core/UserAccountDTO.cs @@ -23,6 +23,7 @@ public UserAccountDto(PostgresUserAccount userAccount) FirstName = userAccount.FirstName; LastName = userAccount.LastName; ClaimedStickerCount = userAccount.ClaimedStickerCount; + PrintedStickerCount = userAccount.PrintedStickerCount; DateCreated = userAccount.DateCreated; } @@ -41,6 +42,9 @@ public UserAccountDto(PostgresUserAccount userAccount) [JsonPropertyName("claimedStickerCount")] public int ClaimedStickerCount { get; set; } + [JsonPropertyName("printedStickerCount")] + public int PrintedStickerCount { get; set; } + [JsonPropertyName("dateCreated")] public DateTime DateCreated { get; set; } } \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubEventPublisher.cs b/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubEventPublisher.cs index 137bc866..a755521e 100644 --- a/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubEventPublisher.cs +++ b/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubEventPublisher.cs @@ -50,7 +50,22 @@ public async Task PublishUserRegisteredEventV1(UserRegisteredEvent userRegistere var formatter = new JsonEventFormatter(); var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); - - await publisherClient.PublishAsync(ByteString.CopyFrom(data.Span)); + + var pubsubMessage = new PubsubMessage + { + Data = ByteString.CopyFrom(data.Span) + }; + + if (Tracer.Instance.ActiveScope is not null) + { + new SpanContextInjector().InjectIncludingDsm( + pubsubMessage.Attributes, + (attrs, key, value) => attrs[key] = value, + Tracer.Instance.ActiveScope.Span.Context, + "googlepubsub", + cloudEvent.Type!); + } + + await publisherClient.PublishAsync(pubsubMessage); } } \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubMessagingWorker.cs b/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubMessagingWorker.cs index 97b6e395..7e48c991 100644 --- a/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubMessagingWorker.cs +++ b/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubMessagingWorker.cs @@ -71,10 +71,23 @@ public async Task StopAsync(CancellationToken cancellationToken) CancellationToken cancellationToken) { logger.LogInformation("GooglePubSubMessagingWorker processing message"); + + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + message.Attributes, + static (attributes, key) => + { + if (attributes.TryGetValue(key, out var value)) + return new[] { value }; + return Enumerable.Empty(); + }, + "googlepubsub", + "users.stickerClaimed.v1"); + // Process the message here var messageText = message.Data.ToStringUtf8(); - using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1"); + using var processSpan = Tracer.Instance.StartActive($"process users.stickerClaimed.v1", + new SpanCreationSettings { Parent = extractedContext }); using var scope = serviceScopeFactory.CreateScope(); var handler = scope.ServiceProvider.GetRequiredService(); diff --git a/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubStickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubStickerPrintedWorker.cs new file mode 100644 index 00000000..b05dba12 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.GCP/GooglePubSubStickerPrintedWorker.cs @@ -0,0 +1,123 @@ +/* + * 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. + */ + +// 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 Datadog, Inc. + +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; +using Datadog.Trace; +using Microsoft.Extensions.Logging; +using Stickerlandia.UserManagement.Core; +using Google.Cloud.PubSub.V1; +using Microsoft.Extensions.DependencyInjection; +using Saunter.Attributes; +using Stickerlandia.UserManagement.Core.Observability; +using Stickerlandia.UserManagement.Core.StickerClaimedEvent; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; + +#pragma warning disable CA1848 + +namespace Stickerlandia.UserManagement.GCP; + +public class GooglePubSubStickerPrintedWorker( + ILogger logger, + IServiceScopeFactory serviceScopeFactory, + [FromKeyedServices("printJobs.completed.v1")] + SubscriberClient subscriber) : IMessagingWorker +{ + private Task? _task; + + public Task StartAsync() + { + logger.LogInformation("GooglePubSubStickerPrintedWorker started"); + + _task = subscriber.StartAsync(ProcessMessageAsync); + + return Task.CompletedTask; + } + + public async Task PollAsync(CancellationToken stoppingToken) + { + try + { + // Run for 5 seconds. + await Task.Delay(5000, stoppingToken); + + if (_task is not null) + { + await _task; + } + } +#pragma warning disable CA1031 + catch (Exception ex) + { + logger.LogWarning(ex, "GooglePubSubStickerPrintedWorker failed"); + } + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + logger.LogInformation("GooglePubSubStickerPrintedWorker stopping"); + + await subscriber.StopAsync(cancellationToken); + } + + [Channel("printJobs.completed.v1")] + [SubscribeOperation(typeof(StickerPrintedEventV1))] + private async Task ProcessMessageAsync(PubsubMessage message, + CancellationToken cancellationToken) + { + logger.LogInformation("GooglePubSubStickerPrintedWorker processing message"); + + var extractedContext = new SpanContextExtractor().ExtractIncludingDsm( + message.Attributes, + static (attributes, key) => + { + if (attributes.TryGetValue(key, out var value)) + return new[] { value }; + return Enumerable.Empty(); + }, + "googlepubsub", + "printJobs.completed.v1"); + + // Process the message here + var messageText = message.Data.ToStringUtf8(); + + using var processSpan = Tracer.Instance.StartActive($"process printJobs.completed.v1", + new SpanCreationSettings { Parent = extractedContext }); + + using var scope = serviceScopeFactory.CreateScope(); + var handler = scope.ServiceProvider.GetRequiredService(); + + var detailBytes = System.Text.Encoding.UTF8.GetBytes(messageText); + var formatter = new JsonEventFormatter(); + var cloudEvent = await formatter.DecodeStructuredModeMessageAsync( + new MemoryStream(detailBytes), null, new List()); + var stickerPrinted = (StickerPrintedEventV1?)cloudEvent.Data; + + if (stickerPrinted == null) + { + processSpan.Span.SetTag("error.message", "Invalid message format"); + return SubscriberClient.Reply.Ack; + } + + try + { + await handler.Handle(stickerPrinted!); + } + catch (InvalidUserException ex) + { + Log.InvalidUser(logger, ex); + + return SubscriberClient.Reply.Nack; + } + + // Acknowledge the message + return SubscriberClient.Reply.Ack; + } +} \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.GCP/ServiceExtensions.cs b/user-management/src/Stickerlandia.UserManagement.GCP/ServiceExtensions.cs index 9531f514..3a958cf8 100644 --- a/user-management/src/Stickerlandia.UserManagement.GCP/ServiceExtensions.cs +++ b/user-management/src/Stickerlandia.UserManagement.GCP/ServiceExtensions.cs @@ -31,7 +31,8 @@ public static IServiceCollection AddGcpAdapters(this IServiceCollection services var projectId = configuration["ConnectionStrings:messaging"]; if (string.IsNullOrEmpty(projectId)) throw new InvalidOperationException("Google ProjectId is not configured."); - services.AddSingleton(); + services.AddKeyedSingleton("stickerClaimed"); + services.AddKeyedSingleton("stickerPrinted"); services.AddKeyedSingleton("users.userRegistered.v1", new PublisherClientBuilder() @@ -45,6 +46,12 @@ public static IServiceCollection AddGcpAdapters(this IServiceCollection services SubscriptionName = new SubscriptionName(projectId, "users.stickerClaimed.v1"), EmulatorDetection = EmulatorDetection.EmulatorOrProduction }.Build()); + services.AddKeyedSingleton("printJobs.completed.v1", + new SubscriberClientBuilder + { + SubscriptionName = new SubscriptionName(projectId, "printJobs.completed.v1"), + EmulatorDetection = EmulatorDetection.EmulatorOrProduction + }.Build()); services.AddSingleton(); diff --git a/user-management/src/Stickerlandia.UserManagement.Lambda/SqsHandler.cs b/user-management/src/Stickerlandia.UserManagement.Lambda/SqsHandler.cs index b6025142..cf6eda67 100644 --- a/user-management/src/Stickerlandia.UserManagement.Lambda/SqsHandler.cs +++ b/user-management/src/Stickerlandia.UserManagement.Lambda/SqsHandler.cs @@ -8,11 +8,14 @@ using Amazon.Lambda.Annotations; using Amazon.Lambda.CloudWatchEvents; using Amazon.Lambda.SQSEvents; +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; using Datadog.Trace; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Stickerlandia.UserManagement.Core; using Stickerlandia.UserManagement.Core.StickerClaimedEvent; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; using Log = Stickerlandia.UserManagement.Core.Observability.Log; namespace Stickerlandia.UserManagement.Lambda; @@ -41,17 +44,23 @@ private async Task ProcessMessage(SQSEvent.SQSMessage message, List failedMessages, StickerClaimedHandler handler) { - var evtData = JsonSerializer.Deserialize>(message.Body, _jsonSerializerOptions); + var evtData = JsonSerializer.Deserialize>(message.Body, _jsonSerializerOptions); if (evtData == null) { failedMessages.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = message.MessageId }); return; } + + var detailBytes = JsonSerializer.SerializeToUtf8Bytes(evtData.Detail, _jsonSerializerOptions); + var formatter = new JsonEventFormatter(); + var cloudEvent = await formatter.DecodeStructuredModeMessageAsync( + new MemoryStream(detailBytes), null, new List()); + var stickerClaimed = (StickerClaimedEventV1?)cloudEvent.Data; try { - await handler.Handle(evtData.Detail!); + await handler.Handle(stickerClaimed!); } catch (InvalidUserException ex) { diff --git a/user-management/src/Stickerlandia.UserManagement.Lambda/StickerPrintedSqsHandler.cs b/user-management/src/Stickerlandia.UserManagement.Lambda/StickerPrintedSqsHandler.cs new file mode 100644 index 00000000..5b8f1dcf --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Lambda/StickerPrintedSqsHandler.cs @@ -0,0 +1,103 @@ +/* + * 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.Text.Json; +using Amazon.Lambda.Annotations; +using Amazon.Lambda.CloudWatchEvents; +using Amazon.Lambda.SQSEvents; +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; +using Datadog.Trace; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Stickerlandia.UserManagement.Core; +using Stickerlandia.UserManagement.Core.RegisterUser; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; +using Log = Stickerlandia.UserManagement.Core.Observability.Log; + +namespace Stickerlandia.UserManagement.Lambda; + +public class StickerPrintedSqsHandler( + ILogger logger, + IServiceScopeFactory serviceScopeFactory) +{ + private readonly JsonSerializerOptions _jsonSerializerOptions = new() { PropertyNameCaseInsensitive = true }; + + [LambdaFunction] + public async Task Handler(SQSEvent sqsEvent) + { + ArgumentNullException.ThrowIfNull(sqsEvent, nameof(sqsEvent)); + + using var scope = serviceScopeFactory.CreateScope(); + var handler = scope.ServiceProvider.GetRequiredService(); + + var failedMessages = new List(); + + foreach (var message in sqsEvent.Records) await ProcessMessage(message, failedMessages, handler); + + return new SQSBatchResponse(failedMessages); + } + + private async Task ProcessMessage(SQSEvent.SQSMessage message, + List failedMessages, + StickerPrintedHandler handler) + { + var evtData = JsonSerializer.Deserialize>(message.Body, _jsonSerializerOptions); + + if (evtData == null) + { + failedMessages.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = message.MessageId }); + return; + } + + // The detail field contains the CloudEvent in structured mode format. + // Re-serialize it to bytes so the CloudEvents formatter can decode it. + var detailBytes = JsonSerializer.SerializeToUtf8Bytes(evtData.Detail, _jsonSerializerOptions); + var formatter = new JsonEventFormatter(); + var cloudEvent = await formatter.DecodeStructuredModeMessageAsync( + new MemoryStream(detailBytes), null, new List()); + var stickerPrinted = (StickerPrintedEventV1?)cloudEvent.Data; + + if (stickerPrinted == null) + { + failedMessages.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = message.MessageId }); + return; + } + + // Extract trace context injected by the publisher. + // Datadog headers (x-datadog-trace-id, etc.) are in the _datadog object at the root of the + // CloudEvent JSON. W3C traceparent is set as a CloudEvent extension attribute. + var propagatedContext = new SpanContextExtractor().ExtractIncludingDsm( + (cloudEvent, evtData.Detail), + static (carrier, key) => + { + var (ce, detail) = carrier; + if (detail.TryGetProperty("_datadog", out var ddObj) && + ddObj.TryGetProperty(key, out var ddVal)) + { + return [ddVal.GetString()]; + } + var ceVal = ce[key]?.ToString(); + return ceVal is null ? [] : (IEnumerable)[ceVal]; + }, + "eventbridge", + cloudEvent.Type!); + + using var processSpan = Tracer.Instance.StartActive( + $"process {cloudEvent.Type}", + new SpanCreationSettings { Parent = propagatedContext }); + + try + { + await handler.Handle(stickerPrinted); + } + catch (InvalidUserException ex) + { + Log.InvalidUser(logger, ex); + failedMessages.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = message.MessageId }); + } + } +} \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Lambda/serverless.template b/user-management/src/Stickerlandia.UserManagement.Lambda/serverless.template index 2f141b98..2f00918e 100644 --- a/user-management/src/Stickerlandia.UserManagement.Lambda/serverless.template +++ b/user-management/src/Stickerlandia.UserManagement.Lambda/serverless.template @@ -70,6 +70,23 @@ "PackageType": "Zip", "Handler": "Stickerlandia.UserManagement.Lambda::Stickerlandia.UserManagement.Lambda.SqsHandler_StickerClaimed_Generated::StickerClaimed" } + }, + "StickerlandiaUserManagementLambdaStickerPrintedSqsHandlerHandlerGenerated": { + "Type": "AWS::Serverless::Function", + "Metadata": { + "Tool": "Amazon.Lambda.Annotations" + }, + "Properties": { + "Runtime": "dotnet10", + "CodeUri": ".", + "MemorySize": 512, + "Timeout": 30, + "Policies": [ + "AWSLambdaBasicExecutionRole" + ], + "PackageType": "Zip", + "Handler": "Stickerlandia.UserManagement.Lambda::Stickerlandia.UserManagement.Lambda.StickerPrintedSqsHandler_Handler_Generated::Handler" + } } }, "Outputs": { diff --git a/user-management/src/Stickerlandia.UserManagement.Worker/Program.cs b/user-management/src/Stickerlandia.UserManagement.Worker/Program.cs index 520fa285..3ed27c82 100644 --- a/user-management/src/Stickerlandia.UserManagement.Worker/Program.cs +++ b/user-management/src/Stickerlandia.UserManagement.Worker/Program.cs @@ -12,6 +12,7 @@ builder.Services.AddHostedService(); builder.Services.AddHostedService(); +builder.Services.AddHostedService(); var host = builder.Build(); host.Run(); \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Worker/StickerClaimedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Worker/StickerClaimedWorker.cs index 56c8c1f9..b483fc73 100644 --- a/user-management/src/Stickerlandia.UserManagement.Worker/StickerClaimedWorker.cs +++ b/user-management/src/Stickerlandia.UserManagement.Worker/StickerClaimedWorker.cs @@ -18,21 +18,15 @@ namespace Stickerlandia.UserManagement.Worker; -internal sealed class StickerClaimedWorker : BackgroundService +internal sealed class StickerClaimedWorker( + ILogger logger, + [FromKeyedServices("stickerClaimed")] IMessagingWorker messagingWorker) + : BackgroundService { - private readonly ILogger _logger; - private readonly IMessagingWorker _messagingWorker; - - public StickerClaimedWorker(ILogger logger, IMessagingWorker messagingWorker) - { - _logger = logger; - _messagingWorker = messagingWorker; - } - protected override async Task ExecuteAsync(CancellationToken stoppingToken) { // Start processing - await _messagingWorker.StartAsync(); + await messagingWorker.StartAsync(); try { @@ -41,13 +35,13 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { try { - await _messagingWorker.PollAsync(stoppingToken); + await messagingWorker.PollAsync(stoppingToken); failureCount = 0; await Task.Delay(1000, stoppingToken); } catch (Exception ex) { - Log.GenericWarning(_logger, "Error processing message", ex); + Log.GenericWarning(logger, "Error processing message", ex); failureCount++; if (failureCount > 10) @@ -59,18 +53,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) } catch (Exception ex) { - Log.GenericWarning(_logger, "Error processing message", ex); + Log.GenericWarning(logger, "Error processing message", ex); } finally { // Stop the processor - await _messagingWorker.StopAsync(stoppingToken); + await messagingWorker.StopAsync(stoppingToken); } } - - public override async Task StopAsync(CancellationToken cancellationToken) - { - await _messagingWorker.StopAsync(cancellationToken); - await base.StopAsync(cancellationToken); - } } \ No newline at end of file diff --git a/user-management/src/Stickerlandia.UserManagement.Worker/StickerPrintedWorker.cs b/user-management/src/Stickerlandia.UserManagement.Worker/StickerPrintedWorker.cs new file mode 100644 index 00000000..a357d5d7 --- /dev/null +++ b/user-management/src/Stickerlandia.UserManagement.Worker/StickerPrintedWorker.cs @@ -0,0 +1,64 @@ +/* + * 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. + */ + +// 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 Datadog, Inc. + +// Allow catch of a generic exception in the worker to ensure the worker failing doesn't crash the entire application. +#pragma warning disable CA1031 +// This is a worker service that is not intended to be instantiated directly, so we suppress the warning. +#pragma warning disable CA1812 + +using Stickerlandia.UserManagement.Core; +using Stickerlandia.UserManagement.Core.Observability; + +namespace Stickerlandia.UserManagement.Worker; + +internal sealed class StickerPrintedWorker( + ILogger logger, + [FromKeyedServices("stickerPrinted")] IMessagingWorker messagingWorker) + : BackgroundService +{ + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + // Start processing + await messagingWorker.StartAsync(); + + try + { + var failureCount = 0; + while (!stoppingToken.IsCancellationRequested) + { + try + { + await messagingWorker.PollAsync(stoppingToken); + failureCount = 0; + await Task.Delay(1000, stoppingToken); + } + catch (Exception ex) + { + Log.GenericWarning(logger, "Error processing message", ex); + failureCount++; + + if (failureCount > 10) + { + throw; + } + } + } + } + catch (Exception ex) + { + Log.GenericWarning(logger, "Error processing message", ex); + } + finally + { + // Stop the processor + await messagingWorker.StopAsync(stoppingToken); + } + } +} \ No newline at end of file diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/AccountTests.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/AccountTests.cs index 8256e365..595c9bee 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/AccountTests.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/AccountTests.cs @@ -64,6 +64,51 @@ public async Task WhenStickerIsClaimedThenAUsersStickerCountShouldIncrement() } } + [Fact] + public async Task WhenStickerIsPrintedThenAUsersPrintedCountShouldIncrement() + { + // Arrange + var emailAddress = $"{Guid.NewGuid()}@test.com"; + var password = $"{Guid.NewGuid()}!A23"; + + // Act + var registerResult = await _driver.RegisterUser(emailAddress, password); + + if (registerResult is null) throw new ArgumentException("Registration failed"); + + var loginResponse = await _driver.Login(emailAddress, password); + + if (loginResponse is null) throw new ArgumentException("Login response is null"); + + var userAccount = await _driver.GetUserAccount(loginResponse.AuthToken); + + await _driver.InjectStickerPrintedMessage(userAccount.AccountId); + + await Task.Delay(TimeSpan.FromSeconds(5)); + + var retryCount = 1; + var maxRetries = 5; + + while (retryCount <= maxRetries) + { + testOutputHelper.WriteLine($"Retry {retryCount} of {maxRetries} to check sticker count..."); + + var user = await _driver.GetUserAccount(loginResponse.AuthToken); + + // Expect the printed sticker count to be 1, break after completed. + if (user!.PrintedStickerCount == 1) + { + break; + } + + retryCount++; + + if (retryCount == maxRetries) Assert.Fail("Failed to increment sticker count after maximum retries."); + + await Task.Delay(TimeSpan.FromSeconds(3)); + } + } + [Fact] public async Task WhenAUserRegistersTheyShouldBeAbleToUpdateTheirDetails() { diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AccountDriver.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AccountDriver.cs index 6cdda63a..8101aca6 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AccountDriver.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AccountDriver.cs @@ -8,7 +8,10 @@ using System.Text; using System.Text.Json; using System.Web; +using CloudNative.CloudEvents; +using CloudNative.CloudEvents.SystemTextJson; using Stickerlandia.UserManagement.Core.StickerClaimedEvent; +using Stickerlandia.UserManagement.Core.StickerPrintedEvent; using Stickerlandia.UserManagement.IntegrationTest.ViewModels; using Xunit.Abstractions; @@ -488,13 +491,39 @@ public async Task InjectStickerClaimedMessage(string userId, string stickerId) { var messagingType = _messaging.GetType(); _testOutputHelper.WriteLine($"Injecting sticker claimed messaging using {messagingType.FullName} messaging."); - - await _messaging.SendMessageAsync("users.stickerClaimed.v1", new StickerClaimedEventV1 + + var messageJson = JsonSerializer.Serialize(new StickerClaimedEventV1 { AccountId = userId, StickerId = stickerId }); - + + await _messaging.SendMessageAsync("users.stickerClaimed.v1", messageJson); + + _testOutputHelper.WriteLine("Injected!"); + } + + public async Task InjectStickerPrintedMessage(string userId) + { + var messagingType = _messaging.GetType(); + _testOutputHelper.WriteLine($"Injecting sticker printed messaging using {messagingType.FullName} messaging."); + + var payload = new StickerPrintedEventV1 { UserId = userId }; + var cloudEvent = new CloudEvent(CloudEventsSpecVersion.V1_0) + { + Id = Guid.NewGuid().ToString(), + Source = new Uri("https://stickerlandia.com"), + Type = "printJobs.completed.v1", + Time = DateTime.UtcNow, + DataContentType = "application/json", + Data = JsonDocument.Parse(JsonSerializer.Serialize(payload)).RootElement + }; + var formatter = new JsonEventFormatter(); + var encoded = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + var messageJson = Encoding.UTF8.GetString(encoded.Span); + + await _messaging.SendMessageAsync("printJobs.completed.v1", messageJson); + _testOutputHelper.WriteLine("Injected!"); } diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AzureServiceBusMessaging.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AzureServiceBusMessaging.cs index 081fdbc8..0ad1dbbb 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AzureServiceBusMessaging.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/AzureServiceBusMessaging.cs @@ -9,7 +9,6 @@ // Copyright 2025 Datadog, Inc. using System.Text; -using System.Text.Json; using Azure.Messaging.ServiceBus; namespace Stickerlandia.UserManagement.IntegrationTest.Drivers; @@ -18,15 +17,15 @@ internal sealed class AzureServiceBusMessaging(string connectionString) : IMessa { private readonly ServiceBusClient _client = new(connectionString); - public async Task SendMessageAsync(string queueName, object message) + public async Task SendMessageAsync(string queueName, string messageJson) { var sender = _client.CreateSender(queueName); - var messageBody = JsonSerializer.Serialize(message); - var serviceBusMessage = new ServiceBusMessage(Encoding.UTF8.GetBytes(messageBody)) + + var serviceBusMessage = new ServiceBusMessage(Encoding.UTF8.GetBytes(messageJson)) { - ContentType = "application/json" + ContentType = "application/cloudevents+json" }; - + await sender.SendMessageAsync(serviceBusMessage); } diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/EventBridgeMessaging.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/EventBridgeMessaging.cs index 2fabe41e..921fc46a 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/EventBridgeMessaging.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/EventBridgeMessaging.cs @@ -8,7 +8,6 @@ // This product includes software developed at Datadog (https://www.datadoghq.com/). // Copyright 2025 Datadog, Inc. -using System.Text.Json; using Amazon.EventBridge; using Amazon.EventBridge.Model; @@ -18,7 +17,7 @@ internal sealed class EventBridgeMessaging(string environment) : IMessaging, IAs { private readonly AmazonEventBridgeClient _client = new(); - public async Task SendMessageAsync(string queueName, object message) + public async Task SendMessageAsync(string queueName, string messageJson) { await _client.PutEventsAsync(new PutEventsRequest { @@ -29,7 +28,7 @@ await _client.PutEventsAsync(new PutEventsRequest EventBusName = $"Stickerlandia-Shared-{environment}", Source = $"{environment}.stickers", DetailType = queueName, - Detail = JsonSerializer.Serialize(message) + Detail = messageJson } } }); diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/GooglePubSubMessaging.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/GooglePubSubMessaging.cs index 5f48e550..606e0a02 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/GooglePubSubMessaging.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/GooglePubSubMessaging.cs @@ -10,20 +10,16 @@ #pragma warning disable CA1812 -using CloudNative.CloudEvents; -using CloudNative.CloudEvents.SystemTextJson; using Google.Api.Gax; using Google.Cloud.PubSub.V1; using Google.Protobuf; using Grpc.Core; -using Stickerlandia.UserManagement.Core.RegisterUser; -using Stickerlandia.UserManagement.Core.StickerClaimedEvent; namespace Stickerlandia.UserManagement.IntegrationTest.Drivers; internal sealed class GooglePubSubMessaging : IMessaging, IAsyncDisposable { - private readonly PublisherClient _client; + private readonly Dictionary _clients = new(); public GooglePubSubMessaging(string connectionString) { @@ -31,65 +27,59 @@ public GooglePubSubMessaging(string connectionString) throw new ArgumentException("Connection string must be provided for Google Pub/Sub messaging.", nameof(connectionString)); - var topicName = new TopicName(connectionString, "users.userRegistered.v1"); - var stickerClaimedTopic = new TopicName(connectionString, "users.stickerClaimed.v1"); - var publisherApiClient = new PublisherServiceApiClientBuilder { EmulatorDetection = EmulatorDetection.EmulatorOrProduction }.Build(); - - try - { - publisherApiClient.CreateTopic(topicName); - Console.WriteLine($"Topic {topicName} created."); - } - catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists) - { - Console.WriteLine($"Topic {topicName} already exists."); - } - - try - { - publisherApiClient.CreateTopic(stickerClaimedTopic); - Console.WriteLine($"Topic {stickerClaimedTopic} created."); - } - catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists) - { - Console.WriteLine($"Topic {stickerClaimedTopic} already exists."); - } - - var subscriber = new SubscriberServiceApiClientBuilder() { EmulatorDetection = EmulatorDetection.EmulatorOrProduction }.Build(); - SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(connectionString, "users.stickerClaimed.v1"); - - try + var subscriber = new SubscriberServiceApiClientBuilder { EmulatorDetection = EmulatorDetection.EmulatorOrProduction }.Build(); + + var topics = new[] { - subscriber.CreateSubscription(subscriptionName, stickerClaimedTopic, pushConfig: null, ackDeadlineSeconds: 60); - } - catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists) + "users.userRegistered.v1", + "users.stickerClaimed.v1", + "printJobs.completed.v1" + }; + + foreach (var topicId in topics) { - // Already exists. That's fine. - } + var topicName = new TopicName(connectionString, topicId); + + try + { + publisherApiClient.CreateTopic(topicName); + Console.WriteLine($"Topic {topicName} created."); + } + catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists) + { + Console.WriteLine($"Topic {topicName} already exists."); + } + + var subscriptionName = SubscriptionName.FromProjectSubscription(connectionString, topicId); + try + { + subscriber.CreateSubscription(subscriptionName, topicName, pushConfig: null, ackDeadlineSeconds: 60); + } + catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists) + { + // Already exists. That's fine. + } - _client = new PublisherClientBuilder { TopicName = stickerClaimedTopic, EmulatorDetection = EmulatorDetection.EmulatorOrProduction}.Build(); + _clients[topicId] = new PublisherClientBuilder + { + TopicName = topicName, + EmulatorDetection = EmulatorDetection.EmulatorOrProduction + }.Build(); + } } - public async Task SendMessageAsync(string queueName, object message) + public async Task SendMessageAsync(string queueName, string messageJson) { - var cloudEvent = new CloudEvent(CloudEventsSpecVersion.V1_0) - { - Id = Guid.NewGuid().ToString(), - Source = new Uri("https://stickerlandia.com"), - Type = queueName, - Time = DateTime.UtcNow, - Data = message - }; - - var formatter = new JsonEventFormatter(); - var data = formatter.EncodeStructuredModeMessage(cloudEvent, out _); + if (!_clients.TryGetValue(queueName, out var client)) + throw new InvalidOperationException($"No publisher client configured for topic '{queueName}'"); - var publishResult = await _client.PublishAsync(ByteString.CopyFrom(data.Span)); + await client.PublishAsync(ByteString.CopyFromUtf8(messageJson)); } public async ValueTask DisposeAsync() { - await _client.DisposeAsync(); + foreach (var client in _clients.Values) + await client.DisposeAsync(); } } \ No newline at end of file diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/IMessaging.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/IMessaging.cs index 410176aa..f58f4284 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/IMessaging.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/IMessaging.cs @@ -14,5 +14,5 @@ namespace Stickerlandia.UserManagement.IntegrationTest.Drivers; public interface IMessaging { - Task SendMessageAsync(string queueName, object message); + Task SendMessageAsync(string queueName, string messageJson); } \ No newline at end of file diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/KafkaMessaging.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/KafkaMessaging.cs index f37494da..e4c9dd9e 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/KafkaMessaging.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/KafkaMessaging.cs @@ -8,7 +8,6 @@ // This product includes software developed at Datadog (https://www.datadoghq.com/). // Copyright 2025 Datadog, Inc. -using System.Text.Json; using Confluent.Kafka; namespace Stickerlandia.UserManagement.IntegrationTest.Drivers; @@ -27,12 +26,15 @@ public KafkaMessaging(string connectionString) Acks = Acks.All }; } - public async Task SendMessageAsync(string queueName, object message) + public async Task SendMessageAsync(string queueName, string messageJson) { using var producer = new ProducerBuilder(config).Build(); - - await producer.ProduceAsync(queueName, new Message { Key = "", Value = JsonSerializer.Serialize(message) }); - + + await producer.ProduceAsync(queueName, new Message + { + Key = "", Value = messageJson + }); + producer.Flush(TimeSpan.FromSeconds(10)); } diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/MockMessaging.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/MockMessaging.cs index f2f18a39..23c57354 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/MockMessaging.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/Drivers/MockMessaging.cs @@ -12,7 +12,7 @@ namespace Stickerlandia.UserManagement.IntegrationTest.Drivers; internal sealed class MockMessaging : IMessaging { - public Task SendMessageAsync(string queueName, object message) + public Task SendMessageAsync(string queueName, string messageJson) { // Mock implementation - just simulate successful message sending return Task.CompletedTask; diff --git a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/ViewModels/UserAccountDTO.cs b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/ViewModels/UserAccountDTO.cs index 1d965017..6f3f08da 100644 --- a/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/ViewModels/UserAccountDTO.cs +++ b/user-management/tests/Stickerlandia.UserManagement.IntegrationTest/ViewModels/UserAccountDTO.cs @@ -30,4 +30,7 @@ internal sealed record UserAccountDTO [JsonPropertyName("claimedStickerCount")] public int ClaimedStickerCount { get; set; } + + [JsonPropertyName("printedStickerCount")] + public int PrintedStickerCount { get; set; } } \ No newline at end of file diff --git a/user-management/tests/Stickerlandia.UserManagement.UnitTest/AccountTests.cs b/user-management/tests/Stickerlandia.UserManagement.UnitTest/AccountTests.cs index 6beeea49..69b55246 100644 --- a/user-management/tests/Stickerlandia.UserManagement.UnitTest/AccountTests.cs +++ b/user-management/tests/Stickerlandia.UserManagement.UnitTest/AccountTests.cs @@ -59,7 +59,7 @@ public async Task GivenAUserAccountExistsShouldBeAbleToUpdateDetails() PostgresUserAccount? capturedAccount = null; var userAccountUnderTest = UserAccount.From(new AccountId(testAccountId), testEmailAddress, - "John", "Doe", 1, DateTime.UtcNow, AccountTier.Std, AccountType.User); + "John", "Doe", 1, 1, DateTime.UtcNow, AccountTier.Std, AccountType.User); using var userManager = CreateFakeUserManager(); A.CallTo(() => userManager.FindByEmailAsync(A.Ignored)) diff --git a/user-management/tests/Stickerlandia.UserManagement.UnitTest/KafkaWorkerLifecycleTests.cs b/user-management/tests/Stickerlandia.UserManagement.UnitTest/KafkaWorkerLifecycleTests.cs new file mode 100644 index 00000000..d1ed1e6e --- /dev/null +++ b/user-management/tests/Stickerlandia.UserManagement.UnitTest/KafkaWorkerLifecycleTests.cs @@ -0,0 +1,148 @@ +/* + * 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 Confluent.Kafka; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Stickerlandia.UserManagement.Agnostic; + +#pragma warning disable CA2007 + +namespace Stickerlandia.UserManagement.UnitTest; + +public class KafkaStickerPrintedWorkerTests +{ + private readonly IConsumer _consumer; + private readonly IKafkaConsumerFactory _factory; + private readonly KafkaStickerPrintedWorker _worker; + + public KafkaStickerPrintedWorkerTests() + { + _consumer = A.Fake>(); + _factory = A.Fake(); + A.CallTo(() => _factory.Create(A.Ignored)).Returns(_consumer); + + _worker = new KafkaStickerPrintedWorker( + A.Fake>(), + A.Fake(), + new ConsumerConfig { BootstrapServers = "localhost:9092", GroupId = "test" }, + new ProducerConfig { BootstrapServers = "localhost:9092" }, + _factory); + } + + [Fact] + public async Task StartAsync_CreatesConsumerAndSubscribesToTopic() + { + await _worker.StartAsync(); + + A.CallTo(() => _factory.Create(A.Ignored)).MustHaveHappenedOnceExactly(); + A.CallTo(() => _consumer.Subscribe("printJobs.completed.v1")).MustHaveHappenedOnceExactly(); + } + + [Fact] + public async Task PollAsync_UsesExistingConsumerWithoutRecreating() + { + A.CallTo(() => _consumer.Consume(A.Ignored)) + .Returns(null!); + + await _worker.StartAsync(); + await _worker.PollAsync(CancellationToken.None); + await _worker.PollAsync(CancellationToken.None); + + // Consumer is only created once across multiple polls + A.CallTo(() => _factory.Create(A.Ignored)).MustHaveHappenedOnceExactly(); + A.CallTo(() => _consumer.Consume(A.Ignored)).MustHaveHappenedTwiceExactly(); + } + + [Fact] + public async Task StopAsync_ClosesConsumer() + { + await _worker.StartAsync(); + await _worker.StopAsync(CancellationToken.None); + + A.CallTo(() => _consumer.Close()).MustHaveHappenedOnceExactly(); + } + + [Fact] + public async Task PollAsync_WhenNoMessage_DoesNotCommit() + { + A.CallTo(() => _consumer.Consume(A.Ignored)) + .Returns(null!); + + await _worker.StartAsync(); + await _worker.PollAsync(CancellationToken.None); + + A.CallTo(() => _consumer.Commit(A>.Ignored)) + .MustNotHaveHappened(); + } +} + +public class KafkaStickerClaimedWorkerTests +{ + private readonly IConsumer _consumer; + private readonly IKafkaConsumerFactory _factory; + private readonly KafkaStickerClaimedWorker _worker; + + public KafkaStickerClaimedWorkerTests() + { + _consumer = A.Fake>(); + _factory = A.Fake(); + A.CallTo(() => _factory.Create(A.Ignored)).Returns(_consumer); + + _worker = new KafkaStickerClaimedWorker( + A.Fake>(), + A.Fake(), + new ConsumerConfig { BootstrapServers = "localhost:9092", GroupId = "test" }, + new ProducerConfig { BootstrapServers = "localhost:9092" }, + _factory); + } + + [Fact] + public async Task StartAsync_CreatesConsumerAndSubscribesToTopic() + { + await _worker.StartAsync(); + + A.CallTo(() => _factory.Create(A.Ignored)).MustHaveHappenedOnceExactly(); + A.CallTo(() => _consumer.Subscribe("users.stickerClaimed.v1")).MustHaveHappenedOnceExactly(); + } + + [Fact] + public async Task PollAsync_UsesExistingConsumerWithoutRecreating() + { + A.CallTo(() => _consumer.Consume(A.Ignored)) + .Returns(null!); + + await _worker.StartAsync(); + await _worker.PollAsync(CancellationToken.None); + await _worker.PollAsync(CancellationToken.None); + + // Consumer is only created once across multiple polls + A.CallTo(() => _factory.Create(A.Ignored)).MustHaveHappenedOnceExactly(); + A.CallTo(() => _consumer.Consume(A.Ignored)).MustHaveHappenedTwiceExactly(); + } + + [Fact] + public async Task StopAsync_ClosesConsumer() + { + await _worker.StartAsync(); + await _worker.StopAsync(CancellationToken.None); + + A.CallTo(() => _consumer.Close()).MustHaveHappenedOnceExactly(); + } + + [Fact] + public async Task PollAsync_WhenNoMessage_DoesNotCommit() + { + A.CallTo(() => _consumer.Consume(A.Ignored)) + .Returns(null!); + + await _worker.StartAsync(); + await _worker.PollAsync(CancellationToken.None); + + A.CallTo(() => _consumer.Commit(A>.Ignored)) + .MustNotHaveHappened(); + } +} diff --git a/user-management/tests/Stickerlandia.UserManagement.UnitTest/StickerPrintedHandlerTests.cs b/user-management/tests/Stickerlandia.UserManagement.UnitTest/StickerPrintedHandlerTests.cs new file mode 100644 index 00000000..cd023d4e --- /dev/null +++ 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(() => handler.Handle(null!)); + } + + [Fact] + public async Task GivenEventWithEmptyUserId_ShouldThrowArgumentException() + { + using var userManager = CreateFakeUserManager(); + var handler = new StickerPrintedHandler(userManager); + + await Assert.ThrowsAsync(() => + handler.Handle(new StickerPrintedEventV1 { UserId = "" })); + } + + [Fact] + public async Task GivenUserDoesNotExist_ShouldThrowInvalidUserException() + { + using var userManager = CreateFakeUserManager(); + A.CallTo(() => userManager.FindByIdAsync(A.Ignored)) + .Returns(Task.FromResult(null)); + + var handler = new StickerPrintedHandler(userManager); + + await Assert.ThrowsAsync(() => + 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(account)); + A.CallTo(() => userManager.UpdateAsync(A.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 CreateFakeUserManager() + { + var userManager = A.Fake>(options => + options.WithArgumentsForConstructor(() => new UserManager( + A.Fake>(), + A.Fake>(), + A.Fake>(), + new List>(), + new List>(), + A.Fake(), + A.Fake(), + CreateFakeServiceProvider(), + A.Fake>>() + ))); + + return userManager; + } + + private static IServiceProvider CreateFakeServiceProvider() + { + var serviceProvider = A.Fake(); + A.CallTo(() => serviceProvider.GetService(typeof(IMeterFactory))) + .Returns(null); + return serviceProvider; + } +} diff --git a/user-management/tests/Stickerlandia.UserManagement.UnitTest/Stickerlandia.UserManagement.UnitTest.csproj b/user-management/tests/Stickerlandia.UserManagement.UnitTest/Stickerlandia.UserManagement.UnitTest.csproj index 51ac5b7d..d8d1874d 100644 --- a/user-management/tests/Stickerlandia.UserManagement.UnitTest/Stickerlandia.UserManagement.UnitTest.csproj +++ b/user-management/tests/Stickerlandia.UserManagement.UnitTest/Stickerlandia.UserManagement.UnitTest.csproj @@ -7,7 +7,7 @@ false true - $(NoWarn);CS0168;CA1031 + $(NoWarn);CS0168;CA1031;CA1707 @@ -31,6 +31,7 @@ +