From 8cc147d367d7c94fdd7610cb0aab88a8124ec40d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 20:35:59 +0100 Subject: [PATCH] Add node-level reactive document shortcuts --- src/GameCult.Mesh/CultMesh.cs | 372 +++++++++++++++--- .../CultMeshStreamingTests.cs | 67 ++++ 2 files changed, 377 insertions(+), 62 deletions(-) diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 45694e6a..3875e251 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -4,6 +4,7 @@ using System.IO; using System.Linq; using System.Net; using System.Net.Sockets; +using System.Reflection; using System.Threading.Tasks; using GameCult.Caching; using GameCult.Caching.MessagePack; @@ -133,35 +134,35 @@ namespace GameCult.Mesh /// Gets or sets the message id prefix used for snapshot requests. public string? MessageIdPrefix { get; set; } - /// Gets or sets schema ids to request. Empty means the document type schema is requested. - public IReadOnlyList? SchemaIds { get; set; } - - /// Gets or sets the target shard id, when the endpoint is shard-aware. - public string? ShardId { get; set; } - - /// Gets or sets the target shard epoch, when the endpoint is shard-aware. - public long? ShardEpoch { get; set; } - + /// Gets or sets schema ids to request. Empty means the document type schema is requested. + public IReadOnlyList? SchemaIds { get; set; } + + /// Gets or sets the target shard id, when the endpoint is shard-aware. + public string? ShardId { get; set; } + + /// Gets or sets the target shard epoch, when the endpoint is shard-aware. + public long? ShardEpoch { get; set; } + /// Gets or sets client security options for endpoint-created clients. public ClientSecurityOptions? Security { get; set; } /// Gets or sets a callback used to configure endpoint-created LiteNetLib clients. public Action? ConfigureClient { get; set; } - - /// Gets or sets the runtime id used by endpoint-created RUDP clients. - public string? RudpRuntimeId { get; set; } - - /// Gets or sets the connection id used by endpoint-created RUDP clients. - public uint RudpConnectionId { get; set; } = 0x43554c54; - - /// Gets or sets the connect payload used by endpoint-created RUDP clients. - public string RudpConnectPayload { get; set; } = "cultnet-schema-rudp"; - - /// Gets or sets the maximum fragment size used by endpoint-created RUDP clients. - public int RudpMaxFragmentBytes { get; set; } = 1024; - - /// Gets or sets the resend delay used by endpoint-created RUDP clients. - public long RudpResendDelayMs { get; set; } = 25; + + /// Gets or sets the runtime id used by endpoint-created RUDP clients. + public string? RudpRuntimeId { get; set; } + + /// Gets or sets the connection id used by endpoint-created RUDP clients. + public uint RudpConnectionId { get; set; } = 0x43554c54; + + /// Gets or sets the connect payload used by endpoint-created RUDP clients. + public string RudpConnectPayload { get; set; } = "cultnet-schema-rudp"; + + /// Gets or sets the maximum fragment size used by endpoint-created RUDP clients. + public int RudpMaxFragmentBytes { get; set; } = 1024; + + /// Gets or sets the resend delay used by endpoint-created RUDP clients. + public long RudpResendDelayMs { get; set; } = 25; } /// @@ -337,6 +338,114 @@ namespace GameCult.Mesh return _host.FlushAsync(soft); } + /// + /// Creates a typed document handle over one distributed record hosted by this node. + /// + public CultMeshDocumentHandle Document( + CultRecordKey key, + CultMeshVerseContext context, + string? documentId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + return CultMesh.Document(Database, key, context, documentId, sources, routeHint); + } + + /// + /// Creates a typed document handle over one distributed record hosted by this node. + /// + public CultMeshDocumentHandle Document( + CultRecordKey key, + CultMeshVerse verse, + string? documentId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return Document(key, verse.Context, documentId, sources, routeHint); + } + + /// + /// Creates a typed document handle over one distributed record hosted by this node. + /// + public CultMeshDocumentHandle Document( + string recordKey, + CultMeshVerseContext context, + string? documentId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + return Document(new CultRecordKey(recordKey), context, documentId, sources, routeHint); + } + + /// + /// Creates a typed document handle over one distributed record hosted by this node. + /// + public CultMeshDocumentHandle Document( + string recordKey, + CultMeshVerse verse, + string? documentId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return Document(recordKey, verse.Context, documentId, sources, routeHint); + } + + /// + /// Creates a managed reactive document mirror over one distributed record hosted by this node. + /// + public Task> ReactiveDocumentAsync( + CultRecordKey key, + CultMeshVerseContext context, + CultMeshReactiveDocumentOptions? options = null) + where TDocument : class + { + return Document(key, context).ReactiveAsync(options); + } + + /// + /// Creates a managed reactive document mirror over one distributed record hosted by this node. + /// + public Task> ReactiveDocumentAsync( + CultRecordKey key, + CultMeshVerse verse, + CultMeshReactiveDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return ReactiveDocumentAsync(key, verse.Context, options); + } + + /// + /// Creates a managed reactive document mirror over one distributed record hosted by this node. + /// + public Task> ReactiveDocumentAsync( + string recordKey, + CultMeshVerseContext context, + CultMeshReactiveDocumentOptions? options = null) + where TDocument : class + { + return ReactiveDocumentAsync(new CultRecordKey(recordKey), context, options); + } + + /// + /// Creates a managed reactive document mirror over one distributed record hosted by this node. + /// + public Task> ReactiveDocumentAsync( + string recordKey, + CultMeshVerse verse, + CultMeshReactiveDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return ReactiveDocumentAsync(recordKey, verse.Context, options); + } + /// public void Dispose() { @@ -1457,17 +1566,15 @@ namespace GameCult.Mesh return Document( ResolveDocumentId(documentId, key), context, - async _ => await ReadRequiredAsync(database, key).ConfigureAwait(false), - _ => database.WatchRecord(key) - .Where(change => change.Document != null) - .Select(change => change.Document!), + async _ => await ReadDatabaseDocumentRequiredAsync(database, key).ConfigureAwait(false), + _ => WatchDatabaseRecordAs(database, key), async value => { - await database.PutAsync(key, value).ConfigureAwait(false); + await PutDatabaseDocumentAsync(database, key, value, predicted: false).ConfigureAwait(false); }, async value => { - await database.PutPredictedAsync(key, value).ConfigureAwait(false); + await PutDatabaseDocumentAsync(database, key, value, predicted: true).ConfigureAwait(false); }, sourceList, route); @@ -2159,44 +2266,44 @@ namespace GameCult.Mesh where TDocument : class { var descriptor = CultDocumentRegistry.Shared.GetRequired(); - return await FetchSnapshotAsync( - createClient, - endpoint, - ToSnapshotRequestOptions( - options, - context, - options.SchemaIds ?? new[] { descriptor.SchemaId }, - new[] { recordKey })) + return await FetchSnapshotAsync( + createClient, + endpoint, + ToSnapshotRequestOptions( + options, + context, + options.SchemaIds ?? new[] { descriptor.SchemaId }, + new[] { recordKey })) .ConfigureAwait(false); } - private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions( + private static CultMeshSnapshotRequestOptions ToSnapshotRequestOptions( CultMeshPeerSnapshotDocumentOptions options, - CultMeshQueryContext context, - IReadOnlyList? schemaIds, - IReadOnlyList? recordKeys) + CultMeshQueryContext context, + IReadOnlyList? schemaIds, + IReadOnlyList? recordKeys) { - return new CultMeshSnapshotRequestOptions + return new CultMeshSnapshotRequestOptions { - SchemaIds = schemaIds, - RecordKeys = recordKeys, - ShardId = options.ShardId, - ShardEpoch = options.ShardEpoch, - ResponseTimeout = options.ResponseTimeout, - ConnectTimeout = options.ConnectTimeout, - MessageIdPrefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix) - ? $"cultmesh:{context.RuntimeId}:snapshot" - : options.MessageIdPrefix, - Security = options.Security, - ConfigureClient = options.ConfigureClient, - RudpRuntimeId = string.IsNullOrWhiteSpace(options.RudpRuntimeId) - ? context.RuntimeId - : options.RudpRuntimeId, - RudpConnectionId = options.RudpConnectionId, - RudpConnectPayload = options.RudpConnectPayload, - RudpMaxFragmentBytes = options.RudpMaxFragmentBytes, - RudpResendDelayMs = options.RudpResendDelayMs - }; + SchemaIds = schemaIds, + RecordKeys = recordKeys, + ShardId = options.ShardId, + ShardEpoch = options.ShardEpoch, + ResponseTimeout = options.ResponseTimeout, + ConnectTimeout = options.ConnectTimeout, + MessageIdPrefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix) + ? $"cultmesh:{context.RuntimeId}:snapshot" + : options.MessageIdPrefix, + Security = options.Security, + ConfigureClient = options.ConfigureClient, + RudpRuntimeId = string.IsNullOrWhiteSpace(options.RudpRuntimeId) + ? context.RuntimeId + : options.RudpRuntimeId, + RudpConnectionId = options.RudpConnectionId, + RudpConnectPayload = options.RudpConnectPayload, + RudpMaxFragmentBytes = options.RudpMaxFragmentBytes, + RudpResendDelayMs = options.RudpResendDelayMs + }; } private static TDocument ReadDocumentFromSnapshotResponse( @@ -2643,5 +2750,146 @@ namespace GameCult.Mesh ?? throw new KeyNotFoundException( $"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}."); } + + private static async Task ReadDatabaseDocumentRequiredAsync( + CultNetDatabase database, + CultRecordKey key) + where TDocument : class + { + var document = await database.GetAsync(key).ConfigureAwait(false); + if (document != null) + return document; + + var untyped = database.Cache.Get(key); + if (untyped != null && IsSameCultDocumentSchema(untyped.GetType())) + return ConvertUntypedDocument(untyped); + + throw new KeyNotFoundException( + $"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}."); + } + + private static R3.Observable WatchDatabaseRecordAs( + CultNetDatabase database, + CultRecordKey key) + where TDocument : class + { + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + return database.WatchAllChanges() + .Where(change => TryConvertDatabaseChange(change, key, descriptor, out TDocument? _)) + .Select(change => + { + TryConvertDatabaseChange(change, key, descriptor, out TDocument? document); + return document!; + }); + } + + private static bool TryConvertDatabaseChange( + object change, + CultRecordKey key, + CultDocumentDescriptor descriptor, + out TDocument? document) + where TDocument : class + { + document = null; + if (change == null) return false; + + var changeType = change.GetType(); + if (changeType.GetProperty("Key")?.GetValue(change) is not CultRecordKey changedKey || + !changedKey.Equals(key)) + { + return false; + } + + var current = changeType.GetProperty("Document")?.GetValue(change); + if (current == null) + return false; + + var schemaId = changeType.GetProperty("SchemaId")?.GetValue(change) as string; + if (!IsSameCultDocumentSchema(current.GetType(), descriptor, schemaId)) + return false; + + document = current as TDocument ?? ConvertUntypedDocument(current); + return true; + } + + private static bool IsSameCultDocumentSchema(Type documentType) + where TDocument : class + { + return IsSameCultDocumentSchema( + documentType, + CultDocumentRegistry.Shared.GetRequired()); + } + + private static bool IsSameCultDocumentSchema( + Type documentType, + CultDocumentDescriptor descriptor, + string? schemaId = null) + { + if (!string.IsNullOrWhiteSpace(schemaId) && + string.Equals(schemaId, descriptor.SchemaId, StringComparison.Ordinal)) + { + return true; + } + + CultDocumentDescriptor storedDescriptor; + try + { + storedDescriptor = CultDocumentRegistry.Shared.GetRequired(documentType); + } + catch (Exception) + { + return false; + } + + return string.Equals(storedDescriptor.SchemaName, descriptor.SchemaName, StringComparison.Ordinal) && + string.Equals(storedDescriptor.SchemaVersion, descriptor.SchemaVersion, StringComparison.Ordinal); + } + + private static TDocument ConvertUntypedDocument(object document) + where TDocument : class + { + var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType()); + return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TDocument), payload); + } + + private static object ConvertUntypedDocument(object document, Type targetType) + { + var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType()); + return CultDocumentMessagePackSerialization.DeserializeUntyped(targetType, payload); + } + + private static async Task PutDatabaseDocumentAsync( + CultNetDatabase database, + CultRecordKey key, + TDocument value, + bool predicted) + where TDocument : class + { + var stored = database.Cache.Get(key); + object outgoing = value; + var outgoingType = typeof(TDocument); + if (stored != null && + !outgoingType.IsInstanceOfType(stored) && + IsSameCultDocumentSchema(stored.GetType(), CultDocumentRegistry.Shared.GetRequired())) + { + outgoing = ConvertUntypedDocument(value, stored.GetType()); + outgoingType = stored.GetType(); + } + + var method = typeof(CultNetDatabase) + .GetMethods(BindingFlags.Instance | BindingFlags.Public) + .Where(candidate => candidate.Name == (predicted + ? nameof(CultNetDatabase.PutPredictedAsync) + : nameof(CultNetDatabase.PutAsync))) + .Single(candidate => + candidate.IsGenericMethodDefinition && + candidate.GetParameters() is { Length: 2 } parameters && + parameters[0].ParameterType == typeof(CultRecordKey)); + + var task = (Task)method + .MakeGenericMethod(outgoingType) + .Invoke(database, new object[] { key, outgoing })!; + await task.ConfigureAwait(false); + } } } diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 2cfa18b8..a13cd9b7 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -1397,6 +1397,73 @@ public sealed class CultMeshStreamingTests cache.Get(key)!.Text.Should().Be("pilot-coalesced-order"); } + [Test] + public async Task NodeReactiveDocument_OpensSameSchemaAliasWithOneCall() + { + var schemaId = CultDocumentRegistry.Shared.GetRequired().SchemaId; + using var temp = new TemporaryDirectory(); + using var node = await CultMesh.CreateNodeAsync( + Path.Combine(temp.Path, "node-reactive-alias.ccmp"), + new CultMeshNodeOptions + { + StartServer = false, + DatabaseOptions = new CultNetDatabaseOptions + { + RuntimeId = "pilot-a", + Shards = new[] + { + new CultNetShardDescriptor( + "pilot-inputs", + "server", + epoch: 1, + isPrimary: false, + schemaIds: new[] { schemaId }, + keyPrefix: "input:") + }, + ClientAuthorityScopes = new[] + { + new CultNetClientAuthorityScope( + "pilot-a", + schemaIds: new[] { schemaId }, + keyPrefix: "input:pilot-a") + } + } + }); + var key = new CultRecordKey("input:pilot-a:node-reactive"); + await node.Cache.UpsertAsync(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "node-initial", + Revision = 1 + }, new CultRecordHandle(key)); + + using var pilotReactive = await node.ReactiveDocumentAsync( + key, + CultMesh.Verse("starbridge", "pilot-a"), + new CultMeshReactiveDocumentOptions + { + FlushDelay = TimeSpan.FromMinutes(1) + }); + var commanderHandle = node.Document( + key.Value, + CultMesh.Verse("starbridge", "commander-rts")); + var observedByCommander = new List(); + using var commanderSubscription = commanderHandle.Watch(observedByCommander.Add); + + pilotReactive.Update(document => + { + document.Text = "node-one-call-prediction"; + document.Revision = 2; + }); + await pilotReactive.FlushAsync(); + + await WaitForAsync(() => observedByCommander.Any(document => document.Text == "node-one-call-prediction")); + pilotReactive.Document.DocumentId.Should().Be(key.Value); + pilotReactive.Document.Context.RuntimeId.Should().Be("pilot-a"); + node.Cache.Get(key)!.Text.Should().Be("node-one-call-prediction"); + (await commanderHandle.AsSchemaAlias().LatestAsync()).Revision.Should().Be(2); + } + [Test] public async Task ReactiveDocument_SyncedSnapshotEndpointCapturesCanonicalCorrectionAcrossRuntimeBoundary() {