[codex] Scope JetStream dedupe by subject

docs/architecture/audit-follow-up-notes.md

@@ -50,7 +50,7 @@ These notes capture architectural and developer-experience findings from the bro
 - Hardened outbox publishing so logger scope/write failures cannot prevent failed publish attempts from being marked failed for retry.
 - Hardened NATS JetStream publishing so post-ack success logging and already-existing stream debug logging cannot turn successful broker operations into false publish failures.
 - Hardened NATS consumer lifecycle/error logging so disabled/no-subscription startup, polling errors, deserialization failures, handler failures, and stream-exists diagnostics cannot change consumer control flow.
-- Added stable NATS JetStream publish de-duplication by using the outbox message id as `MsgId` and treating duplicate broker acks as successful idempotent publishes.
+- Added stable NATS JetStream publish de-duplication with a subject-scoped digest of the normalized subject and outbox message id, and treated duplicate broker acks as successful idempotent publishes.
 - Added `Unknown = 0` to public module contract and domain-state enums, guarded it with an architecture test, and fixed Catalog/Ordering mapping so unknown status no longer becomes active/orderable state.
 - Hardened `eng/new-module.ps1` so `-RegisterInHost` fails loudly when the expected host-registration anchor is missing and prints explicit follow-up steps for `ArchitectureCatalog`, host composition, and enum validation conventions.
 - Replaced Catalog read-model enum casts with explicit switch mapping and added an architecture guard against direct casts to public module enums outside generated migrations.

docs/architecture/messaging-and-outbox.md

@@ -54,10 +54,14 @@ The worker can publish only module outbox stores registered in that worker proce
 Lower-level test or custom hosts can still reference `Gma.Framework.Messaging.Nats`, provide `INatsConnection` themselves, and call `AddNatsJetStreamMessaging()` directly, but production hosts should use the configured Aspire adapter so connection-string behavior stays consistent.
 The low-level messaging methods compose `AddMessagingInfrastructure()` idempotently, and that composes `AddRuntimeInfrastructure()` for shared clocks and id generation without pulling in CQRS or domain-event dispatch.
 
-The NATS JetStream adapter publishes each outbox row with the outbox message id as `NatsJSPubOpts.MsgId`.
+The NATS JetStream adapter publishes each outbox row with a fixed-length `NatsJSPubOpts.MsgId`: the lowercase SHA-256 digest of the normalized subject plus the outbox message id. Scoping broker de-duplication to both values preserves idempotent retries for one subject without suppressing a different event subject that deliberately shares the same logical event id.
 If the broker accepted a message but the local outbox mark-processed step failed, a later retry may publish the same outbox row again. JetStream duplicate tracking then returns a duplicate ack instead of storing another message, and the adapter treats that ack as a successful idempotent publish.
 Consumers must still keep inbox idempotency because delivery remains at-least-once.
 
+The subject-scoped identifier replaced the earlier GUID-only broker identifier. During an upgrade, a row first published by the earlier adapter and retried by the updated adapter can be stored once under each identifier; drain the publishing backlog before rollout when that overlap is unacceptable, and always retain consumer inbox idempotency.
+
+This broker-level subject scoping does not change module outbox storage: each module outbox table remains keyed by its GUID message id. Producers must keep integration-event ids globally collision-resistant because two rows with the same id cannot coexist in one module outbox even when their subjects differ.
+
 ## Subject Format
 
 Integration event subjects use:

src/Messaging/Gma.Framework.Messaging.Nats/NatsJetStreamEventBus.cs

@@ -1,14 +1,15 @@
 namespace Gma.Framework.Messaging.Nats;
 
+using System.Security.Cryptography;
 using System.Text;
+using Gma.Framework.Messaging;
+using Gma.Framework.Runtime;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using Microsoft.Extensions.Options;
 using NATS.Client.Core;
 using NATS.Client.JetStream;
 using NATS.Client.JetStream.Models;
-using Gma.Framework.Messaging;
-using Gma.Framework.Runtime;
 
 #pragma warning disable IDE0290 // Explicit constructor selects the shared-manager DI path over the compatibility overload.
 public sealed class NatsJetStreamEventBus : IEventBus, IDisposable
@@ -53,7 +54,7 @@ public async Task PublishAsync(OutboxMessageRecord message, CancellationToken ca
         byte[] payload = Encoding.UTF8.GetBytes(message.Payload);
         NatsJSPubOpts publishOptions = new()
         {
-            MsgId = CreateMessageId(message.Id)
+            MsgId = CreateMessageId(message.Subject, message.Id)
         };
         PubAckResponse ack = await jetStream
             .PublishAsync(message.Subject, payload, opts: publishOptions, cancellationToken: cancellationToken)
@@ -69,8 +70,18 @@ public async Task PublishAsync(OutboxMessageRecord message, CancellationToken ca
         this.LogPublished(message.Subject);
     }
 
-    private static string CreateMessageId(Guid messageId) =>
-        messageId.ToString("N");
+    internal static string CreateMessageId(string subject, Guid messageId)
+    {
+        if (messageId == Guid.Empty)
+        {
+            throw new ArgumentException("messageId must not be empty.", nameof(messageId));
+        }
+
+        string normalizedSubject = IntegrationEventNaming.NormalizeSubject(subject);
+        byte[] identity = Encoding.UTF8.GetBytes(
+            string.Concat(normalizedSubject, "\0", messageId.ToString("N")));
+        return Convert.ToHexStringLower(SHA256.HashData(identity));
+    }
 
     public void Dispose() => this.ownedStreamManager?.Dispose();
 

tests/Gma.Framework.Tests/Gma.Framework.Tests.csproj

@@ -9,6 +9,7 @@
     <PackageReference Include="Microsoft.NET.Test.Sdk" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
     <PackageReference Include="System.CommandLine" />
+    <PackageReference Include="Testcontainers" />
     <PackageReference Include="Testcontainers.MsSql" />
     <PackageReference Include="Testcontainers.PostgreSql" />
     <PackageReference Include="xunit" />

tests/Gma.Framework.Tests/Messaging/EventBusTests.cs

@@ -1,13 +1,13 @@
 namespace Gma.Framework.Tests;
 
 using System.Reflection;
-using Microsoft.Extensions.Logging.Abstractions;
-using Microsoft.Extensions.Options;
-using NATS.Client.Core;
 using Gma.Framework.Messaging;
-using Gma.Framework.Messaging.Nats;
 using Gma.Framework.Messaging.Infrastructure;
+using Gma.Framework.Messaging.Nats;
 using Gma.Framework.Runtime;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using NATS.Client.Core;
 using Xunit;
 
 [Trait("Category", "Unit")]
@@ -69,6 +69,45 @@ public async Task Null_event_bus_reports_missing_adapter_for_real_messages()
         Assert.Contains("No integration event bus is configured", exception.Message, StringComparison.Ordinal);
     }
 
+    [Fact]
+    public void Nats_message_id_is_fixed_length_deterministic_and_subject_scoped()
+    {
+        Guid messageId = Guid.Parse("d46d078a-bb29-4c4f-979d-f0da1ca8b40b");
+
+        string canonical = NatsJetStreamEventBus.CreateMessageId(
+            "gma.catalog.item-created.v1",
+            messageId);
+        string normalizedEquivalent = NatsJetStreamEventBus.CreateMessageId(
+            " GMA.CATALOG.ITEM-CREATED.V1 ",
+            messageId);
+        string differentSubject = NatsJetStreamEventBus.CreateMessageId(
+            "gma.catalog.item-updated.v1",
+            messageId);
+        string differentMessage = NatsJetStreamEventBus.CreateMessageId(
+            "gma.catalog.item-created.v1",
+            Guid.Parse("8f8c1025-9444-4c30-960a-6fe5bd2ceabc"));
+
+        Assert.Equal(
+            "0b8c00a025b2c478ee5813f38909f1e3ded7fcad6bbfcca6c1cfb9d289bde2c8",
+            canonical);
+        Assert.Equal(64, canonical.Length);
+        Assert.Matches("^[0-9a-f]{64}$", canonical);
+        Assert.Equal(canonical, normalizedEquivalent);
+        Assert.NotEqual(canonical, differentSubject);
+        Assert.NotEqual(canonical, differentMessage);
+    }
+
+    [Fact]
+    public void Nats_message_id_rejects_invalid_identity_inputs()
+    {
+        Assert.Throws<ArgumentException>(() => NatsJetStreamEventBus.CreateMessageId(
+            "gma.catalog.item-created.v1",
+            Guid.Empty));
+        Assert.ThrowsAny<ArgumentException>(() => NatsJetStreamEventBus.CreateMessageId(
+            "catalog.item-created",
+            Guid.NewGuid()));
+    }
+
     private static INatsConnection CreateUnusedNatsConnection() =>
         DispatchProxy.Create<INatsConnection, UnusedNatsConnectionProxy>();
 

tests/Gma.Framework.Tests/Messaging/NatsJetStreamEventBusProviderTests.cs

@@ -0,0 +1,105 @@
+namespace Gma.Framework.Tests.Messaging;
+
+using DotNet.Testcontainers.Builders;
+using DotNet.Testcontainers.Containers;
+using Gma.Framework.Messaging;
+using Gma.Framework.Messaging.Nats;
+using Gma.Framework.Runtime;
+using Gma.Framework.Tests.Support;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using NATS.Client.Core;
+using NATS.Client.JetStream;
+using Xunit;
+
+[Trait("Category", "Integration")]
+[Trait("Category", "Docker")]
+public sealed class NatsJetStreamEventBusProviderTests
+{
+    private const int NatsPort = 4222;
+
+    [DockerFact]
+    public async Task Dedupe_is_subject_scoped_and_expires_with_the_broker_window()
+    {
+        await using IContainer nats = CreateNatsContainer();
+        await nats.StartAsync();
+
+        TimeSpan duplicateWindow = TimeSpan.FromSeconds(1);
+        string streamName = $"GMA_DEDUPE_{Guid.NewGuid():N}".ToUpperInvariant();
+        await using NatsConnection connection = new(new NatsOpts
+        {
+            Url = GetNatsConnectionString(nats),
+        });
+        using NatsJetStreamStreamManager manager = new(
+            connection,
+            Options.Create(new NatsJetStreamOptions
+            {
+                StreamName = streamName,
+                Storage = NatsStreamStorage.Memory,
+                DuplicateWindow = duplicateWindow,
+            }),
+            Options.Create(new ApplicationIdentityOptions { Namespace = "gma" }),
+            NullLogger<NatsJetStreamStreamManager>.Instance);
+        using NatsJetStreamEventBus eventBus = new(
+            connection,
+            manager,
+            NullLogger<NatsJetStreamEventBus>.Instance);
+        Guid sharedMessageId = Guid.NewGuid();
+        OutboxMessageRecord created = CreateMessage(
+            sharedMessageId,
+            "gma.catalog.item-created.v1",
+            "Gma.Framework.Tests.ItemCreatedIntegrationEvent",
+            "created");
+        OutboxMessageRecord updated = CreateMessage(
+            sharedMessageId,
+            "gma.catalog.item-updated.v1",
+            "Gma.Framework.Tests.ItemUpdatedIntegrationEvent",
+            "updated");
+
+        await eventBus.PublishAsync(created, CancellationToken.None);
+        await eventBus.PublishAsync(created, CancellationToken.None);
+        Assert.Equal(1L, await GetStoredMessageCountAsync(connection, streamName));
+
+        await eventBus.PublishAsync(updated, CancellationToken.None);
+        Assert.Equal(2L, await GetStoredMessageCountAsync(connection, streamName));
+
+        await Task.Delay(duplicateWindow + TimeSpan.FromSeconds(1));
+        await eventBus.PublishAsync(created, CancellationToken.None);
+        Assert.Equal(3L, await GetStoredMessageCountAsync(connection, streamName));
+    }
+
+    private static IContainer CreateNatsContainer() =>
+        new ContainerBuilder("nats:2.10-alpine")
+            .WithPortBinding(NatsPort, assignRandomHostPort: true)
+            .WithCommand("-js")
+            .WithWaitStrategy(Wait.ForUnixContainer().UntilInternalTcpPortIsAvailable(NatsPort))
+            .Build();
+
+    private static string GetNatsConnectionString(IContainer container) =>
+        $"nats://localhost:{container.GetMappedPublicPort(NatsPort)}";
+
+    private static async Task<long> GetStoredMessageCountAsync(
+        INatsConnection connection,
+        string streamName)
+    {
+        NatsJSContext jetStream = new(connection);
+        INatsJSStream stream = await jetStream.GetStreamAsync(
+            streamName,
+            cancellationToken: CancellationToken.None);
+        return stream.Info.State.Messages;
+    }
+
+    private static OutboxMessageRecord CreateMessage(
+        Guid id,
+        string subject,
+        string eventType,
+        string suffix) =>
+        new(
+            id,
+            subject,
+            eventType,
+            1,
+            "tenant-a",
+            DateTimeOffset.UtcNow,
+            $$"""{"suffix":"{{suffix}}"}""");
+}