diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 5eefdaba..1f6e665f 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -140,6 +140,27 @@ namespace GameCult.Mesh public Action? ConfigureClient { get; set; } } + /// + /// Options for opening one typed document from a CultCache backing-store publication. + /// + public sealed class CultMeshStoreDocumentOptions + { + /// Gets or sets the semantic document id. Defaults to the record key. + public string? DocumentId { get; set; } + + /// Gets or sets the source id advertised by diagnostics. Defaults to the record key. + public string? SourceId { get; set; } + + /// Gets or sets the route hint for the resulting handle. + public CultMeshRouteHint? RouteHint { get; set; } + + /// Gets or sets the polling interval for watch fallback. + public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250); + + /// Gets or sets the document registry used to resolve persisted schema aliases. + public CultDocumentRegistry? Registry { get; set; } + } + /// /// A locally hosted CultMesh node over CultCache, CultNet transport, and the mesh database facade. /// @@ -1144,6 +1165,106 @@ namespace GameCult.Mesh routeHint); } + /// + /// Creates a read-only typed document handle over one CultCache backing-store publication. + /// + public static CultMeshDocumentHandle DocumentFromStore( + CacheBackingStore store, + CultRecordKey key, + CultMeshVerseContext context, + CultMeshStoreDocumentOptions? options = null) + where TDocument : class + { + if (store == null) throw new ArgumentNullException(nameof(store)); + if (context == null) throw new ArgumentNullException(nameof(context)); + + var resolvedOptions = options ?? new CultMeshStoreDocumentOptions(); + var cache = new CultCache(resolvedOptions.Registry ?? CultDocumentRegistry.Shared); + cache.AddBackingStore(store); + + var descriptor = cache.Registry.GetRequired(); + var documentId = ResolveDocumentId(resolvedOptions.DocumentId, key); + var route = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.InProcess, "CultCache backing store"); + var sources = new[] + { + ProjectionSource( + string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? key.Value : resolvedOptions.SourceId!, + descriptor.SchemaId, + "CultCache backing store") + }; + Func> latest = async _ => + { + await cache.PullAllBackingStoresAsync().ConfigureAwait(false); + return ReadRequired(cache, key); + }; + var watch = PollingQueryWatcher( + async (_parameters, queryContext) => await latest(queryContext).ConfigureAwait(false), + new CultMeshPollingWatchOptions(resolvedOptions.PollInterval)); + + return Document( + documentId, + context, + latest, + queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext), + sources, + route); + } + + /// + /// Creates a read-only typed document handle over one CultCache backing-store publication. + /// + public static CultMeshDocumentHandle DocumentFromStore( + CacheBackingStore store, + CultRecordKey key, + CultMeshVerse verse, + CultMeshStoreDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return DocumentFromStore(store, key, verse.Context, options); + } + + /// + /// Creates a read-only typed document handle over one single-file MessagePack CultCache publication. + /// + public static CultMeshDocumentHandle DocumentFromSingleFile( + string path, + CultRecordKey key, + CultMeshVerseContext context, + CultMeshStoreDocumentOptions? options = null) + where TDocument : class + { + if (string.IsNullOrWhiteSpace(path)) throw new ArgumentException("Value must be non-empty.", nameof(path)); + var resolvedOptions = options ?? new CultMeshStoreDocumentOptions(); + var fileOptions = new CultMeshStoreDocumentOptions + { + DocumentId = resolvedOptions.DocumentId, + SourceId = resolvedOptions.SourceId, + RouteHint = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.SharedMemory, path), + PollInterval = resolvedOptions.PollInterval, + Registry = resolvedOptions.Registry + }; + return DocumentFromStore( + new SingleFileMessagePackBackingStore(path), + key, + context, + fileOptions); + } + + /// + /// Creates a read-only typed document handle over one single-file MessagePack CultCache publication. + /// + public static CultMeshDocumentHandle DocumentFromSingleFile( + string path, + CultRecordKey key, + CultMeshVerse verse, + CultMeshStoreDocumentOptions? options = null) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return DocumentFromSingleFile(path, key, verse.Context, options); + } + /// /// Creates a typed document handle directly over one distributed CultNet database record. /// diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 77ebb014..c03e63d7 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -166,6 +166,23 @@ using var sub = input.Watch() }); ``` +Read-only CultCache publications can be lifted into the same document handle +shape without local store plumbing: + +```csharp +var health = CultMesh.DocumentFromSingleFile( + "daemon-publication.ccmp", + new CultRecordKey("daemon:aetheria.health.v1"), + verse); + +using var healthSub = health.Watch().Subscribe(RenderHealth); +var latestHealth = await health.LatestAsync(); +``` + +`DocumentFromStore()` accepts any `CacheBackingStore`; both helpers pull, +decode, resolve persisted schema metadata, and poll for updates behind the +`CultMeshDocumentHandle` surface. + ## Game Session `CultMeshGameSession` wires the common gameplay loop: local catalogs, diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 9729d555..b753e86a 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -849,6 +849,63 @@ public sealed class CultMeshStreamingTests (await handle.LatestAsync()).Text.Should().Be("predicted-thermal"); } + [Test] + public async Task DocumentHandle_ReadsSchemaPublicationsFromSingleFileStores() + { + using var directory = new TemporaryDirectory(); + var filePath = Path.Combine(directory.Path, "publication.ccmp"); + var key = new CultRecordKey("mesh-note:publication"); + + var writerCache = new CultCache(); + writerCache.AddBackingStore(new SingleFileMessagePackBackingStore(filePath)); + await writerCache.UpsertAsync(typeof(MeshPublicationNoteDocument), new MeshPublicationNoteDocument + { + Schema = "tests.mesh_publication_note.v1", + Text = "published", + Revision = 1 + }, key); + writerCache.FlushAllBackingStores(); + File.Exists(filePath).Should().BeTrue(); + + var directReadCache = new CultCache(); + directReadCache.AddBackingStore(new SingleFileMessagePackBackingStore(filePath)); + await directReadCache.PullAllBackingStoresAsync(); + directReadCache.Get(key)!.Text.Should().Be("published"); + + var handle = CultMesh.DocumentFromSingleFile( + filePath, + key, + CultMesh.Verse("starbridge", "unity-pilot"), + new CultMeshStoreDocumentOptions + { + DocumentId = "daemon:tests.mesh_note.latest", + SourceId = "daemon:tests.mesh_note.latest.v1", + PollInterval = TimeSpan.FromMilliseconds(10) + }); + var observed = new List(); + using var subscription = handle.Watch(value => observed.Add(value)); + + (await handle.LatestAsync()).Text.Should().Be("published"); + handle.CanReplace.Should().BeFalse(); + handle.CanSubmitPrediction.Should().BeFalse(); + handle.DocumentId.Should().Be("daemon:tests.mesh_note.latest"); + handle.RouteHint.Kind.Should().Be(CultMeshLocalityKind.SharedMemory); + handle.Sources.Should().ContainSingle().Which.SourceId.Should().Be("daemon:tests.mesh_note.latest.v1"); + + var republisher = new CultCache(); + republisher.AddBackingStore(new SingleFileMessagePackBackingStore(filePath)); + await republisher.UpsertAsync(typeof(MeshPublicationNoteDocument), new MeshPublicationNoteDocument + { + Schema = "tests.mesh_publication_note.v1", + Text = "republished", + Revision = 2 + }, key); + republisher.FlushAllBackingStores(); + + await WaitForAsync(() => observed.Any(value => value.Text == "republished")); + (await handle.LatestAsync()).Revision.Should().Be(2); + } + [Test] public async Task DocumentHandle_ReadsRemotePeerSnapshotsAsTypedDocuments() { @@ -1393,6 +1450,22 @@ public sealed class CultMeshStreamingTests return catalog; } + private static async Task WaitForAsync(Func predicate) + { + var deadline = DateTimeOffset.UtcNow.AddSeconds(2); + while (DateTimeOffset.UtcNow < deadline) + { + if (predicate()) + { + return; + } + + await Task.Delay(20); + } + + predicate().Should().BeTrue(); + } + private sealed class TemporaryDirectory : IDisposable { public TemporaryDirectory() @@ -1470,6 +1543,20 @@ public sealed class CultMeshStreamingTests public int Revision { get; set; } } + [CultDocument("tests.mesh_publication_note", "tests.mesh_publication_note.v1")] + [MessagePackObject(AllowPrivate = true)] + internal sealed class MeshPublicationNoteDocument + { + [Key(0)] + public string Schema { get; set; } = string.Empty; + + [Key(1)] + public string Text { get; set; } = string.Empty; + + [Key(2)] + public int Revision { get; set; } + } + [CultDocument("tests.mesh_other", "tests.mesh_other.v1")] [MessagePackObject(AllowPrivate = true)] internal sealed class MeshOtherDocument