Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -745,7 +745,7 @@ public static IReadOnlyCollection<EntityUpdate> ToEntityUpdates(
)
{ OldValue = pointer.Evaluate(current) };
})
.DistinctBy(x => new { x.Id, x.Collection })
.DistinctBy(x => new { Id = x.Id is JsonElement je ? je.GetRawText() : x.Id?.ToString(), x.Collection })
.ToArray();

internal static (JsonElement, JsonPatch) UpdateJsonElement<TChange>(this TChange request, JsonElement? currentJson, JsonSerializerOptions options) where TChange : JsonChange
Expand Down Expand Up @@ -991,7 +991,7 @@ public static IReadOnlyCollection<EntityUpdate> ToEntityUpdates(
)
{ OldValue = idSegment == null ? current.Instances : current.Instances.GetValueOrDefault(idSegment) };
})
.DistinctBy(x => new { x.Id, x.Collection })
.DistinctBy(x => new { Id = x.Id is JsonElement je ? je.GetRawText() : x.Id?.ToString(), x.Collection })
.ToArray();

internal static (InstanceCollection, JsonPatch) UpdateJsonElement(this DataChangedEvent request, InstanceCollection current, JsonSerializerOptions options)
Expand Down
21 changes: 17 additions & 4 deletions src/MeshWeaver.Graph/MeshDataSource.cs
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,9 @@ public static MessageHubConfiguration AddMeshDataSource(
return data
.Configure(rm => rm
.ForReducedStream<InstanceCollection>(reduced => reduced
.AddWorkspaceReference<MeshNodeReference, MeshNode>(ReduceToMeshNode))
.AddWorkspaceReference<MeshNodeReference, MeshNode>(
(ci, r, initial) => ReduceToMeshNode(
ci, r, initial, data.Workspace.Hub.JsonSerializerOptions)))
.ForReducedStream<MeshNode>(reduced => reduced
.AddPatchFunction(PatchMeshNode))
.AddWorkspaceReferenceStream<MeshNode>(
Expand Down Expand Up @@ -1125,8 +1127,9 @@ private static IMessageDelivery HandleRunTests(
/// stale snapshot) and instances to bind to the wrong assembly.
/// </para>
/// </summary>
private static ChangeItem<MeshNode> ReduceToMeshNode(
ChangeItem<InstanceCollection> current, MeshNodeReference reference, bool initial)
internal static ChangeItem<MeshNode> ReduceToMeshNode(
ChangeItem<InstanceCollection> current, MeshNodeReference reference, bool initial,
JsonSerializerOptions options)
{
var instances = current.Value?.Instances.Values.OfType<MeshNode>();
var node = !string.IsNullOrEmpty(reference.Path)
Expand All @@ -1152,7 +1155,17 @@ u.Value is MeshNode m
// returning null (which silently drops the emission and blocks live updates).
return new(node, current.StreamId, current.Version);
}
return new(change.Value as MeshNode, current.ChangedBy, current.StreamId,
// change.Value is a JsonElement when the update was derived from a JSON
// patch (cross-hub / mirror path) — `as MeshNode` would silently null it
// out, dropping a null-content ChangeItem into the mirror. Deserialize it
// the same way the sibling PatchMeshNode does.
var changedNode = change.Value switch
{
MeshNode m => m,
JsonElement je => je.Deserialize<MeshNode>(options),
_ => null
};
return new(changedNode, current.ChangedBy, current.StreamId,
ChangeType.Patch, current.Version, [change]);
}

Expand Down
52 changes: 52 additions & 0 deletions test/MeshWeaver.Data.Test/EntityUpdateDedupTest.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
using System.Linq;
using System.Text.Json;
using Json.Patch;
using MeshWeaver.Data.Serialization;
using Xunit;

namespace MeshWeaver.Data.Test;

/// <summary>
/// Deterministic repro + spec for <see cref="JsonSynchronizationStream.ToEntityUpdates(JsonElement, JsonElement, JsonPatch, JsonSerializerOptions, MeshWeaver.Domain.ITypeRegistry?)"/>.
/// <para>
/// Two patch operations that touch DIFFERENT fields of the SAME entity must collapse into a
/// SINGLE <see cref="EntityUpdate"/> (the method's contract: "one distinct update per affected
/// entity, deduplicated by id and collection"). The id is decoded via
/// <see cref="JsonSynchronizationStream.DecodePointerSegment"/> which yields a boxed
/// <see cref="JsonElement"/> — a struct cursor with NO value equality — so the original
/// <c>DistinctBy(x =&gt; new { x.Id, x.Collection })</c> never deduped and leaked duplicate
/// per-entity updates into the sync stream. The fix keys on the JsonElement's raw text.
/// </para>
/// </summary>
public class EntityUpdateDedupTest
{
[Fact]
public void TwoOps_SameEntity_DedupeToOneUpdate()
{
var options = new JsonSerializerOptions();

// Store projection: { "<Collection>": { "<json-encoded id>": { entity } } }.
// The id object key is JSON-encoded ("k1" with quote chars) exactly as the
// pointer segments are encoded — mirrors EncodePointerSegment/DecodePointerSegment.
var current = JsonDocument.Parse(
"""{ "TestType": { "\"k1\"": { "Name": "old", "Count": 1 } } }""").RootElement;
var updated = JsonDocument.Parse(
"""{ "TestType": { "\"k1\"": { "Name": "new", "Count": 5 } } }""").RootElement;

// TWO operations targeting the SAME (collection=TestType, id="k1") but different leaves.
var patch = JsonSerializer.Deserialize<JsonPatch>(
"""
[
{ "op": "replace", "path": "/TestType/\"k1\"/Name", "value": "new" },
{ "op": "replace", "path": "/TestType/\"k1\"/Count", "value": 5 }
]
""", options)!;

var updates = current.ToEntityUpdates(updated, patch, options);

// Pre-fix: the boxed-JsonElement id never compares equal → DistinctBy keeps BOTH (count 2).
// Post-fix: keyed on GetRawText() → the two ops collapse to ONE update.
Assert.Single(updates);
Assert.Equal("TestType", updates.First().Collection);
}
}
47 changes: 47 additions & 0 deletions test/MeshWeaver.Graph.Test/ReduceToMeshNodeJsonElementTest.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
using System.Collections.Generic;
using System.Text.Json;
using MeshWeaver.Data;
using MeshWeaver.Mesh;
using Xunit;

namespace MeshWeaver.Graph.Test;

/// <summary>
/// Deterministic repro + spec for <see cref="MeshDataSourceExtensions.ReduceToMeshNode"/> on the
/// patch path. When an update is derived from a JSON patch (the cross-hub / mirror path), the
/// <see cref="EntityUpdate.Value"/> is a boxed <see cref="JsonElement"/>, NOT a <see cref="MeshNode"/>.
/// The original <c>change.Value as MeshNode</c> silently produced <c>null</c>, pushing a
/// null-content <see cref="ChangeItem{T}"/> into the mirror (a live update that drops the node).
/// The fix deserializes the JsonElement to MeshNode exactly as the sibling <c>PatchMeshNode</c> does.
/// </summary>
public class ReduceToMeshNodeJsonElementTest
{
[Fact]
public void PatchPath_JsonElementValue_DeserializesToMeshNode()
{
var options = new JsonSerializerOptions();
var expected = new MeshNode("k1", "TestSpace") { Name = "Hello" };

// The cross-hub/mirror shape: the EntityUpdate carries the node as a JsonElement.
var asJson = JsonSerializer.SerializeToElement(expected, options);
var update = new EntityUpdate(nameof(MeshNode), expected.Id, asJson)
{
OldValue = expected
};

var change = new ChangeItem<InstanceCollection>(
new InstanceCollection(new Dictionary<object, object> { [expected.Id] = expected }),
ChangedBy: "tester",
StreamId: "stream-1",
ChangeType: ChangeType.Patch,
Version: 1,
Updates: [update]);

var result = MeshDataSourceExtensions.ReduceToMeshNode(
change, new MeshNodeReference(), initial: false, options);

// Pre-fix: `change.Value as MeshNode` → null → a null-content ChangeItem reaches the mirror.
Assert.NotNull(result.Value);
Assert.Equal(expected, result.Value);
}
}
Loading