Live aggregation stamps a strong-typed aggregate id (#426)

CLAUDE.md

@@ -1389,8 +1389,15 @@ reloads an inline `Snapshot<T>`.
   - **The declared id type is checked against the identity member**, because the source generator keys
     the dispatcher on the member's type and a mismatch would fail at the first event, far from the
     registration.
-  - ⚠️ **Live aggregation leaves a strong-typed `Id` at its default** — fisher#426, pre-existing and not
-    AOT-specific (it reproduces under the JIT on 2.80.1). The smoke asserts the folded state only.
+  - **Live aggregation stamps a strong-typed `Id` too, since fisher#426.** `AggregateIdentity.TrySetIdentity`
+    backfills the stream id onto an aggregate whose `Create` did not set it, and it assigned only a value
+    the member could hold as-is — a raw `Guid` is never a `KilnId`, so every wrapper was skipped and came
+    back at its default from live aggregation, `FetchForWriting`, `FetchManyForWriting` and `ProjectLatest`.
+    Not AOT-specific: it reproduced under the JIT. It now wraps through `StrongTypedId.Wrap`, and only a
+    wrapper around the stream id's own type, which is the rule `FetchForWriting<T, TId>` refuses by.
+    `strong_typed_aggregate_identity` fails 5 of 7 against the old backfill. The inline snapshot never had
+    the bug (it goes through the projection's identity setter), and that test pins it. The smoke asserts
+    the ids natively.
 - **The smoke consumer references `JasperFx.Events.SourceGenerator` itself**, because a project
   reference does not carry the analyzer the package bundles. Its version is kept in step by hand.
 

HANDOFF.md

@@ -12,7 +12,7 @@ equivalent for and never will.
 [CLAUDE.md](CLAUDE.md) has the architecture and the SQLite traps. This document is the compliance
 scoreboard and the things that are true right now but not obvious from either.
 
-**2598 tests green on net9.0 and net10.0** — 2531 in `Fisher.Tests`, 36 in
+**2605 tests green on net9.0 and net10.0** — 2538 in `Fisher.Tests`, 36 in
 `Fisher.AspNetCore.Tests` and 31 in `Fisher.EntityFrameworkCore.Tests`. 678 of
 them are shared cross-store compliance tests — 509 event sourcing and 169 document.
 On JasperFx **2.80.2** / Weasel **9.40.0**.

README.md

@@ -27,7 +27,7 @@ process**. There is no server to install, nothing to provision, and nothing to k
 > `Fisher.AspNetCore` and `Fisher.EntityFrameworkCore` companion packages all work and are tested.
 >
 > Fisher passes **all 58 suites and 678 tests** it enrolls from `JasperFx.Events.ComplianceTests`,
-> the shared cross-store suite Marten and Polecat also enroll in, alongside its own 2,531.
+> the shared cross-store suite Marten and Polecat also enroll in, alongside its own 2,538.
 >
 > That suite pins **API portability, not behavioural equivalence** — code written against one store
 > compiles and runs against another. It does not pin that the three behave identically, and they do

smoke/aot-consumer/Program.cs

@@ -163,7 +163,7 @@
     await using (var session = store.LightweightSession())
     {
         var live = await session.Events.AggregateStreamAsync<Pod>(pod);
-        Expect(live is { Bed: "garden", Peas: 1 }, "a strong-typed aggregate folds live");
+        Expect(live is { Bed: "garden", Peas: 1 } && live.Id == new PodId(pod), "a strong-typed aggregate folds live and carries its id");
 
         var stream = await session.Events.FetchForWriting<Pod, PodId>(new PodId(pod));
         Expect(stream.Aggregate is { Peas: 1 } && stream.CurrentVersion == 2, "FetchForWriting by a strong-typed id");
@@ -189,7 +189,7 @@
     await using (var session = store.LightweightSession())
     {
         var stream = await session.Events.FetchForWriting<Shoal>(shoal);
-        Expect(stream.Aggregate is { Fish: 2 },
+        Expect(stream.Aggregate is { Fish: 2 } && stream.Aggregate.Id == new ShoalId(shoal),
             "a live-only strong-typed aggregate folds through its declared identity");
     }
 

src/Fisher.Tests/Events/strong_typed_aggregate_identity.cs

@@ -0,0 +1,158 @@
+using JasperFx;
+using JasperFx.Events;
+using JasperFx.Events.Projections;
+
+namespace Fisher.Tests.Events;
+
+/// <summary>
+///     fisher#426 — a live-aggregated aggregate keyed on a strong-typed id gets the stream's identity.
+/// </summary>
+/// <remarks>
+///     <para>
+///         Live aggregation never consults the aggregate's own id, so an aggregate whose <c>Create</c>
+///         does not set it is backfilled afterwards (<c>AggregateIdentity.TrySetIdentity</c>), as Marten
+///         and Polecat do. That backfill assigned the stream id only when the member's type could hold it
+///         as-is, and a <see cref="Guid" /> is never a <see cref="KilnId" />, so a wrapper was skipped and
+///         the aggregate came back with a default id. The fold was right and the identity was not, with
+///         nothing to say so.
+///     </para>
+///     <para>
+///         Every path that backfills is covered, because they are separate call sites: live aggregation,
+///         <c>FetchForWriting</c> by raw id and by wrapper, <c>FetchManyForWriting</c> and
+///         <c>ProjectLatest</c>. The inline snapshot is covered too, because the issue asked whether it
+///         was stamped, and it goes through the projection's identity setter rather than the backfill.
+/// </para>
+/// </remarks>
+public class strong_typed_aggregate_identity
+{
+    private static CancellationToken Token => TestContext.Current.CancellationToken;
+
+    private static DocumentStore GuidStore(TemporaryDatabase database, Action<StoreOptions>? configure = null)
+        => DocumentStore.For(options =>
+        {
+            options.ConnectionString = database.ConnectionString;
+            options.AutoCreateSchemaObjects = AutoCreate.All;
+            configure?.Invoke(options);
+        });
+
+    private static async Task<Guid> StartKilnAsync(DocumentStore store)
+    {
+        var id = Guid.NewGuid();
+        await using var session = store.LightweightSession();
+        session.Events.StartStream<Kiln>(id, new KilnFired(900), new KilnFired(1100));
+        await session.SaveChangesAsync(Token);
+        return id;
+    }
+
+    [Fact]
+    public async Task live_aggregation_stamps_a_guid_backed_wrapper()
+    {
+        using var database = TemporaryDatabase.Create("strong-live-id");
+        await using var store = GuidStore(database);
+        var id = await StartKilnAsync(store);
+
+        await using var query = store.QuerySession();
+        var kiln = await query.Events.AggregateStreamAsync<Kiln>(id, token: Token);
+
+        kiln.ShouldNotBeNull();
+        kiln.Firings.ShouldBe(2);
+        kiln.Id.ShouldBe(new KilnId(id));
+    }
+
+    [Fact]
+    public async Task fetch_for_writing_stamps_a_wrapper_by_raw_id_and_by_wrapper()
+    {
+        using var database = TemporaryDatabase.Create("strong-fetch-id");
+        await using var store = GuidStore(database);
+        var id = await StartKilnAsync(store);
+
+        await using var session = store.LightweightSession();
+
+        (await session.Events.FetchForWriting<Kiln>(id, Token)).Aggregate!.Id.ShouldBe(new KilnId(id));
+        (await session.Events.FetchForWriting<Kiln, KilnId>(new KilnId(id), Token)).Aggregate!.Id
+            .ShouldBe(new KilnId(id));
+    }
+
+    [Fact]
+    public async Task fetch_many_for_writing_stamps_each_wrapper()
+    {
+        using var database = TemporaryDatabase.Create("strong-fetch-many-id");
+        await using var store = GuidStore(database);
+        var first = await StartKilnAsync(store);
+        var second = await StartKilnAsync(store);
+
+        await using var session = store.LightweightSession();
+        var streams = await session.Events.FetchManyForWriting<Kiln>([first, second], Token);
+
+        streams.Select(x => x.Aggregate!.Id).ShouldBe([new KilnId(first), new KilnId(second)]);
+    }
+
+    [Fact]
+    public async Task project_latest_stamps_a_wrapper()
+    {
+        using var database = TemporaryDatabase.Create("strong-project-latest-id");
+        await using var store = GuidStore(database);
+        var id = await StartKilnAsync(store);
+
+        await using var session = store.LightweightSession();
+        session.Events.Append(id, new KilnFired(1200));
+
+        var projected = await session.Events.ProjectLatest<Kiln>(id, Token);
+
+        projected.ShouldNotBeNull();
+        projected.Firings.ShouldBe(3);
+        projected.Id.ShouldBe(new KilnId(id));
+    }
+
+    [Fact]
+    public async Task live_aggregation_stamps_a_string_backed_wrapper()
+    {
+        using var database = TemporaryDatabase.Create("strong-live-key");
+        await using var store = GuidStore(database, options => options.Events.StreamIdentity = StreamIdentity.AsString);
+
+        await using (var session = store.LightweightSession())
+        {
+            session.Events.StartStream<Kettle>("kettle-1", new KilnFired(100));
+            await session.SaveChangesAsync(Token);
+        }
+
+        await using var query = store.QuerySession();
+        var kettle = await query.Events.AggregateStreamAsync<Kettle>("kettle-1", token: Token);
+
+        kettle!.Id.ShouldBe(new KettleKey("kettle-1"));
+    }
+
+    /// <remarks>
+    ///     The inline snapshot never had the bug — the projection's identity setter wraps through the
+    ///     storage — and this pins that, since the issue could not say either way.
+    /// </remarks>
+    [Fact]
+    public async Task an_inline_snapshot_carries_the_wrapper()
+    {
+        using var database = TemporaryDatabase.Create("strong-snapshot-id");
+        await using var store = GuidStore(database,
+            options => options.Projections.Snapshot<Kiln, KilnId>(SnapshotLifecycle.Inline));
+        var id = await StartKilnAsync(store);
+
+        await using var query = store.QuerySession();
+        var kiln = await query.LoadAsync<Kiln, KilnId>(new KilnId(id), Token);
+
+        kiln!.Id.ShouldBe(new KilnId(id));
+    }
+
+    /// <remarks>
+    ///     A wrapper whose identity is set by <c>Create</c> to the stream id ends up with the same value
+    ///     either way. What must not happen is a wrapper around the other identity type being "converted":
+    ///     the backfill only ever assigns a wrapper around the stream id's own type, as the
+    ///     <c>FetchForWriting&lt;T, TId&gt;</c> refusal does.
+    /// </remarks>
+    [Fact]
+    public void a_wrapper_around_another_type_is_left_alone()
+    {
+        var kettle = new Kettle();
+
+        Fisher.Storage.AggregateIdentity.TrySetIdentity(kettle, Guid.NewGuid());
+
+        kettle.Id.ShouldBe(default);
+    }
+}

src/Fisher/Storage/AggregateIdentity.cs

@@ -115,26 +115,53 @@ internal static Type ResolveIdType(Type aggregateType, StreamIdentity streamIden
     ///     settable identity member of a compatible type.
     /// </summary>
     /// <remarks>
-    ///     Live aggregation folds events without ever consulting the aggregate's own id, so an
-    ///     aggregate whose <c>Create</c> method does not set <c>Id</c> would otherwise come back with a
-    ///     default one. Marten and Polecat both backfill it here rather than in the aggregator.
+    ///     <para>
+    ///         Live aggregation folds events without ever consulting the aggregate's own id, so an
+    ///         aggregate whose <c>Create</c> method does not set <c>Id</c> would otherwise come back with a
+    ///         default one. Marten and Polecat both backfill it here rather than in the aggregator.
+    ///     </para>
+    ///     <para>
+    ///         <b>A strong-typed id is wrapped first</b> (fisher#426). The stream id is a raw Guid or
+    ///         string, which a wrapper member cannot hold as-is, so it used to be skipped and the aggregate
+    ///         came back with a default id from live aggregation, <c>FetchForWriting</c>,
+    ///         <c>FetchManyForWriting</c> and <c>ProjectLatest</c> alike. Only a wrapper around the stream
+    ///         id's own type is built, so a Guid is never turned into a string-backed id or the reverse —
+    ///         the rule <c>FetchForWriting&lt;T, TId&gt;</c> refuses by.
+    ///     </para>
     /// </remarks>
     internal static void TrySetIdentity(object aggregate, object streamId)
     {
         var member = FindIdMember(aggregate.GetType());
 
         switch (member)
         {
-            case PropertyInfo { CanWrite: true } property when property.PropertyType.IsInstanceOfType(streamId):
-                property.SetValue(aggregate, streamId);
+            case PropertyInfo { CanWrite: true } property when IdentityValueFor(property.PropertyType, streamId) is { } value:
+                property.SetValue(aggregate, value);
                 break;
 
-            case FieldInfo { IsInitOnly: false } field when field.FieldType.IsInstanceOfType(streamId):
-                field.SetValue(aggregate, streamId);
+            case FieldInfo { IsInitOnly: false } field when IdentityValueFor(field.FieldType, streamId) is { } value:
+                field.SetValue(aggregate, value);
                 break;
         }
     }
 
+    /// <summary>
+    ///     The stream id as a value an identity member of <paramref name="memberType" /> can hold, or null.
+    /// </summary>
+    private static object? IdentityValueFor(Type memberType, object streamId)
+    {
+        if (memberType.IsInstanceOfType(streamId))
+        {
+            return streamId;
+        }
+
+        var target = Nullable.GetUnderlyingType(memberType) ?? memberType;
+
+        return StrongTypedId.TryResolve(target, out var wrapper) && wrapper.SimpleType == streamId.GetType()
+            ? StrongTypedId.Wrap(wrapper, streamId)
+            : null;
+    }
+
     /// <summary>
     ///     Whether the aggregate type has opted out of single-stream identity with
     ///     <see cref="BoundaryAggregateAttribute" />.