Add MetadataFactory to centralize Metadata construction

Introduces MetadataFactory with 4 Create overloads:
- Full (EventStore read path with all fields)
- From event object (conventions compute streamId + enrichers)
- Explicit fields (tests with known sourceStreamId)
- Generic <TEvent, TId> (conventions + duck-typed Id from event)

PlumberEngine.ReadEventData and EventAggregatorEventHandlerStarter
now use MetadataFactory instead of direct Metadata construction.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rafalmaciag

CLAUDE.md

@@ -44,6 +44,7 @@ MicroPlumberd is a CQRS/Event Sourcing framework built on EventStore that emphas
 #### 1. Event Sourcing Layer
 - **PlumberEngine** serves as the core abstraction over EventStore operations
 - **IPlumber/IPlumberInstance** interfaces provide the main API surface
+- **MetadataFactory** centralizes `Metadata` construction with proper JSON schema (`plumber.MetadataFactory` or standalone `new MetadataFactory()`)
 - Events are stored in streams following naming conventions (e.g., `agg-{type}-{id}` for aggregates)
 - Supports snapshots for aggregate performance optimization
 

README.md

@@ -315,6 +315,24 @@ services.AddPlumberd(configure: c => {
     .AddCommandHandler<FooCommandHandler>()
 ```
 
+### MetadataFactory
+
+`MetadataFactory` centralizes `Metadata` construction with proper JSON schema. Available via `plumber.MetadataFactory` or standalone with `new MetadataFactory()` (uses default conventions).
+
+```csharp
+// From an event object — conventions compute sourceStreamId and enrich metadata
+var metadata = plumber.MetadataFactory.Create(context, evt, recipientId);
+
+// With explicit sourceStreamId — for tests and simple scenarios
+var factory = new MetadataFactory();
+var metadata = factory.Create($"Category-{id}", created: DateTimeOffset.Now, userId: "admin");
+
+// Typed — conventions compute sourceStreamId from event type
+var metadata = factory.Create<OrderCreated, Guid>(orderId, evt, created: DateTimeOffset.Now);
+```
+
+The factory ensures `Metadata.StreamId<T>()`, `Metadata.Created()`, `Metadata.UserId()` and other accessors work correctly by building the proper JSON `Data` schema.
+
 ### Conventions
   - SteamNameConvention - from aggregate type, and aggregate id
   - EventNameConvention - from aggregate? instance and event instance

src/MicroPlumberd.Services.BatchOperations.Tests/BatchOperationModelTests.cs

@@ -1,4 +1,3 @@
-using System.Text.Json;
 using FluentAssertions;
 using MicroPlumberd;
 using Xunit;
@@ -53,28 +52,13 @@ namespace MicroPlumberd.Services.BatchOperations.Tests;
 /// </summary>
 public class BatchOperationModelTests
 {
+    private static readonly MetadataFactory _metadataFactory = new();
+
     /// <summary>
     /// Creates test Metadata with minimal required fields.
     /// </summary>
     private static Metadata CreateMetadata(Guid id, DateTimeOffset? created = null)
-    {
-        var eventId = Guid.NewGuid();
-        var streamId = $"BatchOperations-{id}";
-
-        // Build metadata JSON with Created timestamp if provided
-        var metadataJson = created.HasValue
-            ? JsonDocument.Parse($"{{\"Created\":\"{created.Value:O}\"}}")
-            : JsonDocument.Parse("{}");
-
-        return new Metadata(
-            id: id,
-            eventId: eventId,
-            sourceStreamPosition: 0,
-            linkStreamPosition: null,
-            sourceStreamId: streamId,
-            data: metadataJson.RootElement
-        );
-    }
+        => _metadataFactory.Create($"BatchOperations-{id}", created: created);
 
     [Fact]
     public async Task Given_BatchOperationStarted_CreatesOperation()

src/MicroPlumberd.Services.EventAggregator/EventAggregatorEventHandlerStarter.cs

@@ -1,4 +1,3 @@
-using System.Text.Json;
 using MicroPlumberd;
 using MicroPlumberd.Services;
 using ModelingEvolution.EventAggregator;
@@ -62,11 +61,7 @@ private void SubscribeToEnvelope<TEvent>(IEventAggregator eventAggregator, THand
             {
                 try
                 {
-                    var sourceStreamId = _plumber.Conventions.StreamNameFromEventConvention(
-                        context, typeof(TEvent), envelope.RecipientId);
-                    var metadataObj = _plumber.Conventions.GetMetadata(context, null, envelope.Event!, null);
-                    var data = JsonSerializer.SerializeToElement(metadataObj);
-                    var metadata = new Metadata(Guid.Empty, Guid.NewGuid(), 0, null, sourceStreamId, data);
+                    var metadata = _plumber.MetadataFactory.Create(context, envelope.Event!, envelope.RecipientId);
                     await handler.Handle(metadata, envelope.Event!);
                 }
                 catch (Exception ex)

src/MicroPlumberd/PlumberEngine.cs

@@ -28,6 +28,7 @@ internal PlumberEngine(EventStoreClientSettings settings, PlumberConfig? config
         PersistentSubscriptionClient = new EventStorePersistentSubscriptionsClient(settings);
         ProjectionManagementClient = new EventStoreProjectionManagementClient(settings);
         Conventions = config.Conventions;
+        MetadataFactory = new MetadataFactory(Conventions);
         SerializerFactory = config.SerializerFactory;
         ServiceProvider = config.ServiceProvider;
         _extension = config.Extension; // Shouldn't we make a copy?
@@ -80,6 +81,11 @@ internal PlumberEngine(EventStoreClientSettings settings, PlumberConfig? config
     /// <inheritdoc/>
     public IReadOnlyConventions Conventions { get; }
 
+    /// <summary>
+    /// Gets the metadata factory for creating <see cref="Metadata"/> instances with proper JSON schema.
+    /// </summary>
+    public MetadataFactory MetadataFactory { get; }
+
     /// <summary>
     /// Creates a subscription to a stream starting from the specified position.
     /// </summary>
@@ -1175,17 +1181,15 @@ public static PlumberEngine Create(EventStoreClientSettings? settings = null, Ac
     internal (object, Metadata) ReadEventData(OperationContext context, EventRecord er, EventRecord? eLink, Type t)
     {
         var streamIdSuffix = er.EventStreamId.Substring(er.EventStreamId.IndexOf('-') + 1);
-        if (!Guid.TryParse(streamIdSuffix, out var aggregateId)) 
+        if (!Guid.TryParse(streamIdSuffix, out var aggregateId))
             aggregateId = streamIdSuffix.ToGuid();
 
         var s = Serializer(t);
         var ev = s.Deserialize(context,er.Data.Span, t)!;
         var m = s.ParseMetadata(context, er.Metadata.Span);
 
-        long? linkStreamPosition = eLink?.EventNumber.ToInt64();
-        long sourceStreamPosition = er.EventNumber.ToInt64();
-        
-        var metadata = new Metadata(aggregateId, er.EventId.ToGuid(), sourceStreamPosition, linkStreamPosition, er.EventStreamId, m);
+        var metadata = MetadataFactory.Create(aggregateId, er.EventStreamId, er.EventId.ToGuid(),
+            er.EventNumber.ToInt64(), eLink?.EventNumber.ToInt64(), m);
         return (ev, metadata);
     }
 

src/MicroPlumberd/Serialization/MetadataFactory.cs

@@ -0,0 +1,109 @@
+using System.Dynamic;
+using System.Text.Json;
+using MicroPlumberd.Utils;
+
+namespace MicroPlumberd;
+
+/// <summary>
+/// Factory for creating <see cref="Metadata"/> instances with proper JSON schema.
+/// Centralizes metadata construction that was previously scattered across PlumberEngine, EventAggregator, and tests.
+/// </summary>
+public class MetadataFactory
+{
+    private readonly IReadOnlyConventions _conventions;
+
+    /// <summary>
+    /// Creates a factory with default conventions (Created timestamp, InvocationContext enrichers).
+    /// </summary>
+    public MetadataFactory() : this(new Conventions()) { }
+
+    /// <summary>
+    /// Creates a factory with custom conventions.
+    /// </summary>
+    /// <param name="conventions">The conventions to use for metadata enrichment and stream naming.</param>
+    public MetadataFactory(IReadOnlyConventions conventions) => _conventions = conventions;
+
+    /// <summary>
+    /// Creates metadata from raw EventStore fields.
+    /// Used by PlumberEngine when reading events from EventStore.
+    /// </summary>
+    public Metadata Create(
+        Guid id,
+        string sourceStreamId,
+        Guid eventId,
+        long sourceStreamPosition,
+        long? linkStreamPosition,
+        JsonElement data)
+    {
+        return new Metadata(id, eventId, sourceStreamPosition, linkStreamPosition, sourceStreamId, data);
+    }
+
+    /// <summary>
+    /// Creates metadata from an event object. Conventions compute the sourceStreamId and enrich metadata.
+    /// The <see cref="Metadata.Id"/> is extracted from the event's Id property via duck typing if present.
+    /// Used by EventAggregator and other in-process event sources.
+    /// </summary>
+    public Metadata Create(
+        OperationContext context,
+        object evt,
+        object? id = null,
+        object? customMetadata = null,
+        IAggregate? aggregate = null)
+    {
+        var sourceStreamId = _conventions.StreamNameFromEventConvention(context, evt.GetType(), id);
+        var enriched = _conventions.GetMetadata(context, aggregate, evt, customMetadata);
+        var data = JsonSerializer.SerializeToElement(enriched);
+        var eventId = _conventions.GetEventIdConvention(context, aggregate, evt).ToGuid();
+        IdDuckTyping.Instance.TryGetGuidId(evt, out var metadataId);
+        return new Metadata(metadataId, eventId, 0, null, sourceStreamId, data);
+    }
+
+    /// <summary>
+    /// Creates metadata with explicit field values.
+    /// Useful for tests and simple scenarios where you know the sourceStreamId.
+    /// </summary>
+    public Metadata Create(
+        string sourceStreamId,
+        DateTimeOffset? created = null,
+        Guid? correlationId = null,
+        Guid? causationId = null,
+        string? userId = null)
+    {
+        var data = BuildData(created, correlationId, causationId, userId);
+        return new Metadata(Guid.Empty, Guid.NewGuid(), 0, null, sourceStreamId, data);
+    }
+
+    /// <summary>
+    /// Creates metadata for a specific event type and id. Conventions compute the sourceStreamId.
+    /// The <see cref="Metadata.Id"/> and <see cref="Metadata.EventId"/> are extracted from the event's Id property via duck typing if present.
+    /// </summary>
+    public Metadata Create<TEvent, TId>(
+        TId id,
+        TEvent evt,
+        DateTimeOffset? created = null,
+        Guid? correlationId = null,
+        Guid? causationId = null,
+        string? userId = null)
+    {
+        var context = OperationContext.Create(Flow.Component);
+        var sourceStreamId = _conventions.StreamNameFromEventConvention(context, typeof(TEvent), id);
+        var data = BuildData(created, correlationId, causationId, userId);
+        var eventId = _conventions.GetEventIdConvention(context, null, evt!).ToGuid();
+        IdDuckTyping.Instance.TryGetGuidId(evt!, out var metadataId);
+        return new Metadata(metadataId, eventId, 0, null, sourceStreamId, data);
+    }
+
+    private static JsonElement BuildData(
+        DateTimeOffset? created,
+        Guid? correlationId,
+        Guid? causationId,
+        string? userId)
+    {
+        IDictionary<string, object> obj = new ExpandoObject();
+        if (created.HasValue) obj["Created"] = created.Value;
+        if (correlationId.HasValue) obj["$correlationId"] = correlationId.Value;
+        if (causationId.HasValue) obj["$causationId"] = causationId.Value;
+        if (userId != null) obj["UserId"] = userId;
+        return JsonSerializer.SerializeToElement(obj);
+    }
+}