From 04561a5bd9ea399752b3a392ad9ff397293af08f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 17:28:15 +0100 Subject: [PATCH] Add CultMesh single-file document helpers --- src/GameCult.Mesh/CultMesh.cs | 2 +- .../CultMeshSingleFileDocuments.cs | 339 ++++++++++++++++++ .../CultMeshStreamingTests.cs | 91 +++++ 3 files changed, 431 insertions(+), 1 deletion(-) create mode 100644 src/GameCult.Mesh/CultMeshSingleFileDocuments.cs diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index e2e4de44..6d392918 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -397,7 +397,7 @@ namespace GameCult.Mesh /// /// Public CultMesh entrypoints. /// - public static class CultMesh + public static partial class CultMesh { /// /// Creates a local CultMesh node with a durable CultCache and the canonical MessagePack backing store. diff --git a/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs b/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs new file mode 100644 index 00000000..5e2295df --- /dev/null +++ b/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs @@ -0,0 +1,339 @@ +using System; +using System.Buffers; +using System.IO; +using System.Linq; +using System.Security.Cryptography; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using MessagePack; + +namespace GameCult.Mesh +{ + /// + /// Schema metadata used when publishing one raw document payload as a CultCache single-file snapshot. + /// + public sealed class CultMeshSingleFileDocumentSchema + { + /// + /// Creates schema metadata for a single-file document publication. + /// + public CultMeshSingleFileDocumentSchema(string schemaId, string schemaName, string schemaVersion) + { + if (string.IsNullOrWhiteSpace(schemaId)) throw new ArgumentException("Value must be non-empty.", nameof(schemaId)); + if (string.IsNullOrWhiteSpace(schemaName)) throw new ArgumentException("Value must be non-empty.", nameof(schemaName)); + if (string.IsNullOrWhiteSpace(schemaVersion)) throw new ArgumentException("Value must be non-empty.", nameof(schemaVersion)); + + SchemaId = schemaId; + SchemaName = schemaName; + SchemaVersion = schemaVersion; + } + + /// + /// Gets the content-derived schema identifier. + /// + public string SchemaId { get; } + + /// + /// Gets the stable schema name. + /// + public string SchemaName { get; } + + /// + /// Gets the schema version. + /// + public string SchemaVersion { get; } + + /// + /// Gets or sets an optional canonical schema description. + /// + public string CanonicalSchemaJson { get; set; } = string.Empty; + + /// + /// Gets or sets compatible schema identifiers accepted for this document. + /// + public string[] CompatibleSchemaIds { get; set; } = Array.Empty(); + + internal CultSchemaCatalogEntry ToCatalogEntry(byte[] payload) + { + return new CultSchemaCatalogEntry + { + SchemaId = SchemaId, + SchemaName = SchemaName, + SchemaVersion = SchemaVersion, + ContentHash = StableHash(payload), + CanonicalSchemaJson = CanonicalSchemaJson, + CompatibleSchemaIds = CompatibleSchemaIds + .Append(SchemaId) + .Where(candidate => !string.IsNullOrWhiteSpace(candidate)) + .Distinct(StringComparer.Ordinal) + .ToArray(), + Members = Array.Empty() + }; + } + + private static string StableHash(byte[] payload) + { + if (payload == null || payload.Length == 0) + return "empty"; + + using var sha = SHA256.Create(); + return BitConverter.ToString(sha.ComputeHash(payload)).Replace("-", "", StringComparison.Ordinal).ToLowerInvariant(); + } + } + + public static partial class CultMesh + { + /// + /// Writes one typed document as a canonical single-file MessagePack CultCache snapshot. + /// + public static void WriteSingleFileDocument( + string path, + CultRecordKey key, + TDocument document, + string? storedAt = null, + CultDocumentRegistry? registry = null) + where TDocument : class + { + if (document == null) throw new ArgumentNullException(nameof(document)); + + var resolvedRegistry = registry ?? CultDocumentRegistry.Shared; + var descriptor = resolvedRegistry.GetRequired(); + WriteSingleFileDocumentPayload( + path, + key, + descriptor.ToCatalogEntry(), + storedAt, + CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TDocument))); + } + + /// + /// Reads one typed document from a single-file MessagePack CultCache snapshot. + /// + public static TDocument ReadSingleFileDocument( + string path, + CultRecordKey key, + CultDocumentRegistry? registry = null) + where TDocument : class + { + var resolvedRegistry = registry ?? CultDocumentRegistry.Shared; + var descriptor = resolvedRegistry.GetRequired(); + var payload = ReadSingleFileDocumentPayload(path, key, descriptor.SchemaId); + return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TDocument), payload); + } + + /// + /// Writes one raw document payload as a canonical single-file MessagePack CultCache snapshot. + /// + public static void WriteSingleFileDocumentPayload( + string path, + CultRecordKey key, + CultMeshSingleFileDocumentSchema schema, + string? storedAt, + byte[] payload) + { + if (schema == null) throw new ArgumentNullException(nameof(schema)); + WriteSingleFileDocumentPayload(path, key, schema.ToCatalogEntry(payload), storedAt, payload); + } + + /// + /// Reads one raw document payload from a single-file MessagePack CultCache snapshot. + /// + public static byte[] ReadSingleFileDocumentPayload( + string path, + CultRecordKey key, + string expectedSchemaId) + { + if (string.IsNullOrWhiteSpace(path)) throw new ArgumentException("Value must be non-empty.", nameof(path)); + if (string.IsNullOrWhiteSpace(expectedSchemaId)) throw new ArgumentException("Value must be non-empty.", nameof(expectedSchemaId)); + + var snapshot = ReadSingleFileSnapshot(path); + var record = snapshot.Records.SingleOrDefault(candidate => string.Equals(candidate.Key, key.Value, StringComparison.Ordinal)) + ?? throw new InvalidDataException($"CultCache document '{path}' does not contain record '{key.Value}'."); + + if (!string.Equals(record.SchemaId, expectedSchemaId, StringComparison.Ordinal)) + { + throw new InvalidDataException( + $"CultCache document '{path}' record '{key.Value}' has schema '{record.SchemaId}', expected '{expectedSchemaId}'."); + } + + if (!PublishesSchema(snapshot.SchemaCatalog, expectedSchemaId)) + { + throw new InvalidDataException( + $"CultCache document '{path}' does not publish schema '{expectedSchemaId}' in its catalog."); + } + + return record.Payload; + } + + /// + /// Reads the only raw document payload from a single-file MessagePack CultCache snapshot. + /// + public static byte[] ReadSingleFileDocumentPayload( + string path, + string expectedSchemaId) + { + if (string.IsNullOrWhiteSpace(path)) throw new ArgumentException("Value must be non-empty.", nameof(path)); + if (string.IsNullOrWhiteSpace(expectedSchemaId)) throw new ArgumentException("Value must be non-empty.", nameof(expectedSchemaId)); + + var snapshot = ReadSingleFileSnapshot(path); + if (snapshot.Records.Length != 1) + { + throw new InvalidDataException( + $"CultCache document '{path}' must contain exactly one record."); + } + + var record = snapshot.Records[0]; + if (!string.Equals(record.SchemaId, expectedSchemaId, StringComparison.Ordinal)) + { + throw new InvalidDataException( + $"CultCache document '{path}' has schema '{record.SchemaId}', expected '{expectedSchemaId}'."); + } + + if (!PublishesSchema(snapshot.SchemaCatalog, expectedSchemaId)) + { + throw new InvalidDataException( + $"CultCache document '{path}' does not publish schema '{expectedSchemaId}' in its catalog."); + } + + return record.Payload; + } + + private static void WriteSingleFileDocumentPayload( + string path, + CultRecordKey key, + CultSchemaCatalogEntry catalogEntry, + string? storedAt, + byte[] payload) + { + if (string.IsNullOrWhiteSpace(path)) throw new ArgumentException("Value must be non-empty.", nameof(path)); + if (catalogEntry == null) throw new ArgumentNullException(nameof(catalogEntry)); + if (string.IsNullOrWhiteSpace(catalogEntry.SchemaId)) + throw new ArgumentException("Catalog entry must include a schema id.", nameof(catalogEntry)); + + payload ??= Array.Empty(); + var snapshot = new CultPersistedStoreSnapshot + { + SchemaCatalog = new[] { catalogEntry }, + Records = new[] + { + new CultPersistedRecord + { + Key = key.Value, + SchemaId = catalogEntry.SchemaId, + StoredAt = string.IsNullOrWhiteSpace(storedAt) ? DateTimeOffset.UtcNow.ToString("O") : storedAt!, + Payload = payload + } + } + }; + + var directory = Path.GetDirectoryName(path); + if (!string.IsNullOrWhiteSpace(directory)) + Directory.CreateDirectory(directory); + + WriteFileAtomically(path, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot)); + } + + private static CultPersistedStoreSnapshot ReadSingleFileSnapshot(string path) + { + var bytes = File.ReadAllBytes(path); + try + { + return CultDocumentMessagePackSerialization.DeserializeSnapshot(bytes); + } + catch (Exception ex) when (ex is MessagePackSerializationException or InvalidOperationException) + { + return ReadLegacySingleFileSnapshot(bytes); + } + } + + private static CultPersistedStoreSnapshot ReadLegacySingleFileSnapshot(byte[] bytes) + { + var reader = new MessagePackReader(bytes); + var fieldCount = reader.ReadArrayHeader(); + var snapshot = new CultPersistedStoreSnapshot(); + if (fieldCount > 0) + snapshot.FormatVersion = reader.ReadString() ?? "cultcache.store.v1"; + + if (fieldCount > 1) + { + var catalogCount = reader.ReadArrayHeader(); + snapshot.SchemaCatalog = new CultSchemaCatalogEntry[catalogCount]; + for (var index = 0; index < catalogCount; index++) + snapshot.SchemaCatalog[index] = ReadLegacySchemaCatalogEntry(ref reader); + } + + if (fieldCount > 2) + { + var recordCount = reader.ReadArrayHeader(); + snapshot.Records = new CultPersistedRecord[recordCount]; + for (var index = 0; index < recordCount; index++) + snapshot.Records[index] = ReadLegacyPersistedRecord(ref reader); + } + + for (var index = 3; index < fieldCount; index++) + reader.Skip(); + + return snapshot; + } + + private static CultSchemaCatalogEntry ReadLegacySchemaCatalogEntry(ref MessagePackReader reader) + { + var fieldCount = reader.ReadArrayHeader(); + var entry = new CultSchemaCatalogEntry(); + if (fieldCount > 0) entry.SchemaId = reader.ReadString() ?? string.Empty; + if (fieldCount > 1) entry.SchemaName = reader.ReadString() ?? string.Empty; + if (fieldCount > 2) entry.SchemaVersion = reader.ReadString() ?? string.Empty; + if (fieldCount > 3) entry.CanonicalSchemaJson = reader.ReadString() ?? string.Empty; + if (fieldCount > 4) reader.Skip(); + if (fieldCount > 5) entry.ContentHash = reader.ReadString() ?? string.Empty; + if (fieldCount > 6) + { + var memberCount = reader.ReadArrayHeader(); + for (var index = 0; index < memberCount; index++) + reader.Skip(); + } + + for (var index = 7; index < fieldCount; index++) + reader.Skip(); + + entry.CompatibleSchemaIds = string.IsNullOrWhiteSpace(entry.SchemaId) + ? Array.Empty() + : new[] { entry.SchemaId }; + entry.Members = Array.Empty(); + return entry; + } + + private static CultPersistedRecord ReadLegacyPersistedRecord(ref MessagePackReader reader) + { + var fieldCount = reader.ReadArrayHeader(); + var record = new CultPersistedRecord(); + if (fieldCount > 0) record.Key = reader.ReadString() ?? string.Empty; + if (fieldCount > 1) record.SchemaId = reader.ReadString() ?? string.Empty; + if (fieldCount > 2) record.StoredAt = reader.ReadString() ?? string.Empty; + if (fieldCount > 3) + { + var payload = reader.ReadBytes(); + record.Payload = payload.HasValue ? payload.Value.ToArray() : Array.Empty(); + } + for (var index = 4; index < fieldCount; index++) + reader.Skip(); + return record; + } + + private static bool PublishesSchema(CultSchemaCatalogEntry[] catalog, string schemaId) + { + return catalog.Any(entry => + string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal) || + entry.CompatibleSchemaIds.Any(candidate => string.Equals(candidate, schemaId, StringComparison.Ordinal))); + } + + private static void WriteFileAtomically(string path, byte[] bytes) + { + var tempPath = path + ".tmp"; + File.WriteAllBytes(tempPath, bytes); + if (File.Exists(path)) + File.Delete(path); + File.Move(tempPath, path); + } + } +} diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 84db81eb..ae862f9f 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; using System.Collections.Generic; using System.IO; using System.Linq; @@ -906,6 +907,63 @@ public sealed class CultMeshStreamingTests (await handle.LatestAsync()).Revision.Should().Be(2); } + [Test] + public async Task SingleFileDocumentHelpers_RoundTripTypedDocuments() + { + using var directory = new TemporaryDirectory(); + var filePath = Path.Combine(directory.Path, "single-document.ccmp"); + var key = new CultRecordKey("mesh-note:single-document"); + + CultMesh.WriteSingleFileDocument( + filePath, + key, + new MeshPublicationNoteDocument + { + Schema = "tests.mesh_publication_note.v1", + Text = "published-directly", + Revision = 3 + }, + storedAt: "2026-06-27T12:00:00.0000000Z"); + + var document = CultMesh.ReadSingleFileDocument(filePath, key); + + document.Text.Should().Be("published-directly"); + document.Revision.Should().Be(3); + + var cache = new CultCache(); + cache.AddBackingStore(new SingleFileMessagePackBackingStore(filePath)); + await cache.PullAllBackingStoresAsync(); + cache.Get(key)!.Text.Should().Be("published-directly"); + } + + [Test] + public void SingleFileDocumentHelpers_ReadLegacySingleDocumentSnapshots() + { + using var directory = new TemporaryDirectory(); + var filePath = Path.Combine(directory.Path, "legacy-single-document.ccmp"); + var key = new CultRecordKey("mesh-note:legacy-single-document"); + var descriptor = CultDocumentRegistry.Shared.GetRequired(); + var payload = CultDocumentMessagePackSerialization.Serialize(new MeshPublicationNoteDocument + { + Schema = "tests.mesh_publication_note.v1", + Text = "legacy-published", + Revision = 4 + }); + + File.WriteAllBytes(filePath, WriteLegacySingleDocumentSnapshot( + key.Value, + descriptor.SchemaId, + descriptor.SchemaName, + descriptor.SchemaVersion, + "2026-06-27T12:00:00.0000000Z", + payload)); + + var document = CultMesh.ReadSingleFileDocument(filePath, key); + + document.Text.Should().Be("legacy-published"); + document.Revision.Should().Be(4); + } + [Test] public async Task DocumentHandle_ReadsRemotePeerSnapshotsAsTypedDocuments() { @@ -1438,6 +1496,39 @@ public sealed class CultMeshStreamingTests ring.Stats().UnavoidableCopyCount.Should().Be(1); } + private static byte[] WriteLegacySingleDocumentSnapshot( + string key, + string schemaId, + string schemaName, + string schemaVersion, + string storedAt, + byte[] payload) + { + var buffer = new ArrayBufferWriter(); + var writer = new MessagePackWriter(buffer); + + writer.WriteArrayHeader(3); + writer.Write("cultcache.store.v1"); + writer.WriteArrayHeader(1); + writer.WriteArrayHeader(7); + writer.Write(schemaId); + writer.Write(schemaName); + writer.Write(schemaVersion); + writer.Write("Legacy Aetheria single document"); + writer.Write(storedAt); + writer.Write("legacy-hash"); + writer.WriteArrayHeader(0); + writer.WriteArrayHeader(1); + writer.WriteArrayHeader(4); + writer.Write(key); + writer.Write(schemaId); + writer.Write(storedAt); + writer.Write(payload); + writer.Flush(); + + return buffer.WrittenSpan.ToArray(); + } + private static CultMeshStreamCatalog CatalogWithByteStream() { var catalog = CultMesh.CreateStreamCatalog();