From eb52b560880121cddd104f4550cd7785320ca441 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 15:09:36 +0100 Subject: [PATCH] Add CultMesh collection handles --- src/GameCult.Mesh/CultMesh.cs | 289 ++++++++++++++++++ src/GameCult.Mesh/CultMeshPrimitives.cs | 176 +++++++++++ src/GameCult.Mesh/docs/public-api.md | 21 ++ src/GameCult.Networking/CultNetDatabase.cs | 27 ++ .../CultMeshStreamingTests.cs | 80 +++++ 5 files changed, 593 insertions(+) diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 7742b8ac..37b8b1bd 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -1136,6 +1136,215 @@ namespace GameCult.Mesh return new CultMeshDocumentCatalog(documents); } + /// + /// Creates a typed collection handle over all local CultCache records assignable to the document type. + /// + public static CultMeshCollectionHandle Collection( + CultCache cache, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (cache == null) throw new ArgumentNullException(nameof(cache)); + + var descriptor = cache.Registry.GetRequired(); + var sourceList = sources?.ToArray() + ?? new[] { ProjectionSource(CollectionId(collectionId, descriptor), descriptor.SchemaId, "CultCache collection") }; + var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.InProcess, "CultCache collection"); + return new CultMeshCollectionHandle( + CollectionId(collectionId, descriptor), + () => Task.FromResult>(cache.GetAll().ToArray()), + () => cache.Watch().Select(change => ToCollectionChange(change, descriptor.SchemaId)), + sourceList, + route); + } + + /// + /// Creates a typed collection handle over the local CultCache record with the supplied CultName. + /// + public static CultMeshCollectionHandle CollectionByName( + CultCache cache, + string name, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (cache == null) throw new ArgumentNullException(nameof(cache)); + if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("Value must be non-empty.", nameof(name)); + + var descriptor = cache.Registry.GetRequired(); + var resolvedId = collectionId ?? $"{descriptor.SchemaId}:name:{name}"; + var sourceList = sources?.ToArray() + ?? new[] { ProjectionSource(resolvedId, descriptor.SchemaId, "CultCache named collection") }; + var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.InProcess, "CultCache named collection"); + return new CultMeshCollectionHandle( + resolvedId, + () => Task.FromResult>(Optional(cache.GetByName(name))), + () => cache.Watch() + .Where(change => MatchesName(cache, name, change)) + .Select(change => ToCollectionChange(change, descriptor.SchemaId)), + sourceList, + route); + } + + /// + /// Creates a typed collection handle over local CultCache records matched by an indexed value. + /// + public static CultMeshCollectionHandle CollectionByIndex( + CultCache cache, + string alias, + string value, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (cache == null) throw new ArgumentNullException(nameof(cache)); + if (string.IsNullOrWhiteSpace(alias)) throw new ArgumentException("Value must be non-empty.", nameof(alias)); + if (string.IsNullOrWhiteSpace(value)) throw new ArgumentException("Value must be non-empty.", nameof(value)); + + var descriptor = cache.Registry.GetRequired(); + var resolvedId = collectionId ?? $"{descriptor.SchemaId}:index:{alias}:{value}"; + var sourceList = sources?.ToArray() + ?? new[] { ProjectionSource(resolvedId, descriptor.SchemaId, "CultCache indexed collection") }; + var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.InProcess, "CultCache indexed collection"); + return new CultMeshCollectionHandle( + resolvedId, + () => Task.FromResult>(Optional(cache.GetByIndex(alias, value))), + () => cache.Watch() + .Where(change => MatchesIndex(cache, alias, value, change)) + .Select(change => ToCollectionChange(change, descriptor.SchemaId)), + sourceList, + route); + } + + /// + /// Creates a typed collection handle over all distributed CultNet database records assignable to the document type. + /// + public static CultMeshCollectionHandle Collection( + CultNetDatabase database, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (database == null) throw new ArgumentNullException(nameof(database)); + + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var sourceList = sources?.ToArray() + ?? new[] { ProjectionSource(CollectionId(collectionId, descriptor), descriptor.SchemaId, "CultNet database collection") }; + var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Automatic, "CultNet database collection"); + return new CultMeshCollectionHandle( + CollectionId(collectionId, descriptor), + () => Task.FromResult>(database.GetAll().ToArray()), + () => database.Watch().Select(ToCollectionChange), + sourceList, + route); + } + + /// + /// Creates a typed collection handle over the distributed CultNet database record with the supplied CultName. + /// + public static CultMeshCollectionHandle CollectionByName( + CultNetDatabase database, + string name, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (database == null) throw new ArgumentNullException(nameof(database)); + if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("Value must be non-empty.", nameof(name)); + + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var resolvedId = collectionId ?? $"{descriptor.SchemaId}:name:{name}"; + var sourceList = sources?.ToArray() + ?? new[] { ProjectionSource(resolvedId, descriptor.SchemaId, "CultNet database named collection") }; + var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Automatic, "CultNet database named collection"); + return new CultMeshCollectionHandle( + resolvedId, + () => Task.FromResult>(Optional(database.GetByName(name))), + () => database.WatchByName(name).Select(ToCollectionChange), + sourceList, + route); + } + + /// + /// Creates a typed collection handle over distributed CultNet database records matched by an indexed value. + /// + public static CultMeshCollectionHandle CollectionByIndex( + CultNetDatabase database, + string alias, + string value, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (database == null) throw new ArgumentNullException(nameof(database)); + if (string.IsNullOrWhiteSpace(alias)) throw new ArgumentException("Value must be non-empty.", nameof(alias)); + if (string.IsNullOrWhiteSpace(value)) throw new ArgumentException("Value must be non-empty.", nameof(value)); + + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var resolvedId = collectionId ?? $"{descriptor.SchemaId}:index:{alias}:{value}"; + var sourceList = sources?.ToArray() + ?? new[] { ProjectionSource(resolvedId, descriptor.SchemaId, "CultNet database indexed collection") }; + var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Automatic, "CultNet database indexed collection"); + return new CultMeshCollectionHandle( + resolvedId, + () => Task.FromResult>(Optional(database.GetByIndex(alias, value))), + () => database.WatchByIndex(alias, value).Select(ToCollectionChange), + sourceList, + route); + } + + /// + /// Creates a typed collection handle over all CultMesh node database records assignable to the document type. + /// + public static CultMeshCollectionHandle Collection( + CultMeshNode node, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (node == null) throw new ArgumentNullException(nameof(node)); + return Collection(node.Database, collectionId, sources, routeHint); + } + + /// + /// Creates a typed collection handle over the CultMesh node database record with the supplied CultName. + /// + public static CultMeshCollectionHandle CollectionByName( + CultMeshNode node, + string name, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (node == null) throw new ArgumentNullException(nameof(node)); + return CollectionByName(node.Database, name, collectionId, sources, routeHint); + } + + /// + /// Creates a typed collection handle over CultMesh node database records matched by an indexed value. + /// + public static CultMeshCollectionHandle CollectionByIndex( + CultMeshNode node, + string alias, + string value, + string? collectionId = null, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + where TDocument : class + { + if (node == null) throw new ArgumentNullException(nameof(node)); + return CollectionByIndex(node.Database, alias, value, collectionId, sources, routeHint); + } + /// /// Describes a typed live feed surface for tooling, UI, and diagnostics. /// @@ -1639,6 +1848,86 @@ namespace GameCult.Mesh : documentId!; } + private static string CollectionId(string? collectionId, CultDocumentDescriptor descriptor) + { + return string.IsNullOrWhiteSpace(collectionId) + ? descriptor.SchemaId + : collectionId!; + } + + private static IReadOnlyList Optional(TDocument? document) + where TDocument : class + { + return document == null + ? Array.Empty() + : new[] { document }; + } + + private static bool MatchesName( + CultCache cache, + string name, + CultCacheDocumentChange change) + where TDocument : class + { + var current = cache.GetByName(name); + return (change.Document != null && ReferenceEquals(current, change.Document)) || + (change.PreviousDocument != null && ReferenceEquals(current, change.PreviousDocument)); + } + + private static bool MatchesIndex( + CultCache cache, + string alias, + string value, + CultCacheDocumentChange change) + where TDocument : class + { + var current = cache.GetByIndex(alias, value); + return (change.Document != null && ReferenceEquals(current, change.Document)) || + (change.PreviousDocument != null && ReferenceEquals(current, change.PreviousDocument)); + } + + private static CultMeshCollectionChange ToCollectionChange( + CultCacheDocumentChange change, + string schemaId) + where TDocument : class + { + return new CultMeshCollectionChange( + change.Kind switch + { + CultCacheDocumentChangeKind.Added => CultMeshCollectionChangeKind.Added, + CultCacheDocumentChangeKind.Updated => CultMeshCollectionChangeKind.Updated, + CultCacheDocumentChangeKind.Removed => CultMeshCollectionChangeKind.Removed, + _ => CultMeshCollectionChangeKind.Updated + }, + change.Key, + schemaId, + change.Document, + change.PreviousDocument); + } + + private static CultMeshCollectionChange ToCollectionChange( + CultNetDatabaseChange change) + where TDocument : class + { + return new CultMeshCollectionChange( + change.Kind switch + { + CultNetDatabaseChangeKind.Added => CultMeshCollectionChangeKind.Added, + CultNetDatabaseChangeKind.Updated => CultMeshCollectionChangeKind.Updated, + CultNetDatabaseChangeKind.Removed => CultMeshCollectionChangeKind.Removed, + CultNetDatabaseChangeKind.Predicted => CultMeshCollectionChangeKind.Predicted, + CultNetDatabaseChangeKind.Reconciled => CultMeshCollectionChangeKind.Reconciled, + CultNetDatabaseChangeKind.SchemaMigrated => CultMeshCollectionChangeKind.SchemaMigrated, + CultNetDatabaseChangeKind.Rejected => CultMeshCollectionChangeKind.Rejected, + _ => CultMeshCollectionChangeKind.Updated + }, + change.Key, + change.SchemaId, + change.Document, + change.PreviousDocument, + change.Message); + } + private static TDocument ReadRequired(CultCache cache, CultRecordKey key) where TDocument : class { diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index 22624d86..5152e738 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -1155,6 +1155,182 @@ namespace GameCult.Mesh } } + /// + /// Describes how a CultMesh collection document changed. + /// + public enum CultMeshCollectionChangeKind + { + /// A document was added. + Added, + /// A document was updated. + Updated, + /// A document was removed. + Removed, + /// A local prediction was published before authority accepted it. + Predicted, + /// A predicted document was reconciled with authority. + Reconciled, + /// A document was accepted through schema migration. + SchemaMigrated, + /// A document change was rejected. + Rejected + } + + /// + /// One typed document change observed through a CultMesh collection handle. + /// + public sealed class CultMeshCollectionChange + where TDocument : class + { + /// Creates a collection change. + public CultMeshCollectionChange( + CultMeshCollectionChangeKind kind, + CultRecordKey key, + string schemaId, + TDocument? document, + TDocument? previousDocument, + string? message = null) + { + Kind = kind; + Key = key; + SchemaId = schemaId ?? ""; + Document = document; + PreviousDocument = previousDocument; + Message = message; + } + + /// Gets the change kind. + public CultMeshCollectionChangeKind Kind { get; } + + /// Gets the changed record key. + public CultRecordKey Key { get; } + + /// Gets the schema id of the changed document. + public string SchemaId { get; } + + /// Gets the current document, when present. + public TDocument? Document { get; } + + /// Gets the previous document, when present. + public TDocument? PreviousDocument { get; } + + /// Gets an optional rejection or diagnostic message. + public string? Message { get; } + } + + /// + /// Typed reactive collection handle with schema-aware alias conversion. + /// + public sealed class CultMeshCollectionHandle + where TDocument : class + { + private static readonly CultDocumentDescriptor Descriptor = + CultDocumentRegistry.Shared.GetRequired(); + + private readonly Func>> _latest; + private readonly Func>> _watchChanges; + + /// Creates a collection handle. + public CultMeshCollectionHandle( + string collectionId, + Func>> latest, + Func>> watchChanges, + IEnumerable? sources = null, + CultMeshRouteHint? routeHint = null) + { + CollectionId = string.IsNullOrWhiteSpace(collectionId) + ? throw new ArgumentException("Value must be non-empty.", nameof(collectionId)) + : collectionId; + _latest = latest ?? throw new ArgumentNullException(nameof(latest)); + _watchChanges = watchChanges ?? throw new ArgumentNullException(nameof(watchChanges)); + Sources = sources?.ToArray() ?? Array.Empty(); + RouteHint = routeHint ?? CultMeshRouteHint.Automatic; + } + + /// Gets the semantic collection id. + public string CollectionId { get; } + + /// Gets the CLR document type presented by this collection. + public Type DocumentType => typeof(TDocument); + + /// Gets the stable CultCache schema name. + public string SchemaName => Descriptor.SchemaName; + + /// Gets the stable CultCache schema version. + public string SchemaVersion => Descriptor.SchemaVersion; + + /// Gets the content-derived schema identifier. + public string SchemaId => Descriptor.SchemaId; + + /// Gets the preferred or observed route for collection access. + public CultMeshRouteHint RouteHint { get; } + + /// Gets typed state sources this collection handle depends on, when known. + public IReadOnlyList Sources { get; } + + /// Reads one coherent collection snapshot. + public Task> LatestAsync() + { + return _latest(); + } + + /// Watches typed collection changes. + public Observable> WatchChanges() + { + return _watchChanges(); + } + + /// Subscribes to typed collection changes. + public IDisposable WatchChanges(Action> onNext) + { + if (onNext == null) throw new ArgumentNullException(nameof(onNext)); + return WatchChanges().Subscribe(onNext); + } + + /// Creates a same-schema alias presentation for another CLR document type. + public CultMeshCollectionHandle AsSchemaAlias() where TAlias : class + { + var aliasDescriptor = CultDocumentRegistry.Shared.GetRequired(); + if (!string.Equals(Descriptor.SchemaName, aliasDescriptor.SchemaName, StringComparison.Ordinal) || + !string.Equals(Descriptor.SchemaVersion, aliasDescriptor.SchemaVersion, StringComparison.Ordinal)) + { + throw new InvalidOperationException( + $"Document type {typeof(TAlias).FullName} uses schema '{aliasDescriptor.SchemaName}' " + + $"version '{aliasDescriptor.SchemaVersion}', but collection '{CollectionId}' exposes " + + $"schema '{SchemaName}' version '{SchemaVersion}'."); + } + + return new CultMeshCollectionHandle( + CollectionId, + async () => (await LatestAsync().ConfigureAwait(false)) + .Select(ConvertDocument) + .ToArray(), + () => WatchChanges().Select(change => new CultMeshCollectionChange( + change.Kind, + change.Key, + change.SchemaId, + change.Document == null ? null : ConvertDocument(change.Document), + change.PreviousDocument == null ? null : ConvertDocument(change.PreviousDocument), + change.Message)), + Sources, + RouteHint); + } + + private static TTarget ConvertDocument(TSource document) + where TSource : class + where TTarget : class + { + if (document == null) throw new ArgumentNullException(nameof(document)); + if (document is TTarget alreadyTyped) + { + return alreadyTyped; + } + + var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TSource)); + return (TTarget)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TTarget), payload); + } + } + /// /// Inspectable metadata for a typed live feed surface. /// diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index e789f8d6..3a35d7f8 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -176,6 +176,13 @@ portable machinery that makes it feel native. `CultMeshDocumentHandle.AsSchemaAlias()`. Domain facades can expose `Document()`, `DocumentBySchema(...)`, `LatestAsync()`, and `Watch()` without owning schema dictionaries or alias serializers. +- `CultMeshCollectionHandle` is the typed reactive edge for + multi-record state. It exposes one `LatestAsync()` collection snapshot and a + `WatchChanges()` stream while hiding whether the backing source is an + in-process `CultCache`, a distributed `CultNetDatabase`, or a `CultMeshNode`. + Collection handles can represent all records of a document type, a named + document view, or an indexed value view, and they support the same + same-schema CLR alias conversion as document handles. - `CultMeshProjectionRecipe` names a reusable projection from typed source state into derived state. It records source handles, route hints, and projection execution, and can be exposed as a typed query @@ -273,6 +280,20 @@ var uiDocking = documents.Document(); var bySchema = documents.DocumentBySchema("gamecult.aetheria.station_refit.v1"); ``` +Typed collections use the same one-call style: + +```csharp +var allies = CultMesh.CollectionByIndex( + node, + "Faction", + "au"); + +var currentAllies = await allies.LatestAsync(); +using var changes = allies.WatchChanges(change => UpdateRoster(change)); + +var uiAllies = allies.AsSchemaAlias(); +``` + ```ts const queryVerse = verse.withRoute("shared-memory", statePath); const commandVerse = verse diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 83c34b82..560f80ab 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -686,6 +686,33 @@ namespace GameCult.Networking return Task.FromResult(_cache.Get(key)); } + /// + /// Gets all documents assignable to the requested type. + /// + public IEnumerable GetAll() where T : class + { + ThrowIfDisposed(); + return _cache.GetAll(); + } + + /// + /// Gets a typed document by its CultName value. + /// + public T? GetByName(string name) where T : class + { + ThrowIfDisposed(); + return _cache.GetByName(name); + } + + /// + /// Gets a typed document by an indexed value. + /// + public T? GetByIndex(string alias, string value) where T : class + { + ThrowIfDisposed(); + return _cache.GetByIndex(alias, value); + } + /// /// Opens a reactive POCO presentation for one distributed CultCache document. /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index eb902889..d3229bdb 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -771,6 +771,71 @@ public sealed class CultMeshStreamingTests observed.Revision.Should().Be(2); } + [Test] + public async Task CollectionHandle_ReadsAndWatchesDatabaseCollectionsByIndex() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + var teamId = Guid.NewGuid().ToString("D"); + var collection = CultMesh.CollectionByIndex( + database, + "TeamId", + teamId); + var changes = new List>(); + using var subscription = collection.WatchChanges(change => changes.Add(change)); + + await database.PutAsync( + new CultRecordKey("player:one"), + new MeshIndexedPlayer + { + Name = "One", + TeamId = teamId, + Score = 10 + }); + await database.PutAsync( + new CultRecordKey("player:two"), + new MeshIndexedPlayer + { + Name = "Two", + TeamId = Guid.NewGuid().ToString("D"), + Score = 20 + }); + + collection.CollectionId.Should().Contain(teamId); + collection.SchemaName.Should().Be("tests.mesh_indexed_player"); + collection.RouteHint.Kind.Should().Be(CultMeshLocalityKind.Automatic); + collection.Sources.Should().ContainSingle().Which.SchemaId.Should().Be(collection.SchemaId); + + var snapshot = await collection.LatestAsync(); + snapshot.Should().ContainSingle().Which.Name.Should().Be("One"); + changes.Should().ContainSingle(); + changes[0].Kind.Should().Be(CultMeshCollectionChangeKind.Added); + changes[0].Document!.Name.Should().Be("One"); + } + + [Test] + public async Task CollectionHandle_SupportsSameSchemaAliases() + { + var cache = new CultCache(); + await cache.UpsertAsync( + new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "collection-alias", + Revision = 4 + }, + new CultRecordHandle(new CultRecordKey("mesh-note:collection"))); + + var collection = CultMesh.Collection(cache); + var alias = collection.AsSchemaAlias(); + + alias.CollectionId.Should().Be(collection.CollectionId); + alias.SchemaName.Should().Be(collection.SchemaName); + + var snapshot = await alias.LatestAsync(); + snapshot.Should().ContainSingle().Which.Text.Should().Be("collection-alias"); + } + [Test] public async Task DocumentHandle_ReplacesThroughCultMeshNodeDatabase() { @@ -1193,6 +1258,21 @@ public sealed class CultMeshStreamingTests public int Health; } + [CultDocument("tests.mesh_indexed_player", "tests.mesh_indexed_player.v1")] + [MessagePackObject(AllowPrivate = true)] + internal sealed class MeshIndexedPlayer + { + [Key(0)] + public string Name { get; set; } = string.Empty; + + [Key(1)] + [CultIndex] + public string TeamId { get; set; } = string.Empty; + + [Key(2)] + public int Score { get; set; } + } + [CultDocument("tests.mesh_note", "tests.mesh_note.v1")] [MessagePackObject(AllowPrivate = true)] internal sealed class MeshNoteDocument