Add typed snapshot sync helpers

This commit is contained in:
Meta Vens 2026-06-27 17:50:02 +01:00
commit 39e13fdb3c
2 changed files with 99 additions and 2 deletions

View file

@ -83,6 +83,32 @@ namespace GameCult.Mesh
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
}
/// <summary>
/// Result from syncing a remote snapshot into a local CultMesh node.
/// </summary>
public sealed class CultMeshSnapshotSyncResult
{
internal CultMeshSnapshotSyncResult(
CultNetSnapshotResponseRawMessage snapshot,
IReadOnlyList<object> appliedDocuments)
{
Snapshot = snapshot ?? throw new ArgumentNullException(nameof(snapshot));
AppliedDocuments = appliedDocuments ?? throw new ArgumentNullException(nameof(appliedDocuments));
}
/// <summary>Gets the raw snapshot that was fetched from the endpoint.</summary>
public CultNetSnapshotResponseRawMessage Snapshot { get; }
/// <summary>Gets the documents applied into the local node cache.</summary>
public IReadOnlyList<object> AppliedDocuments { get; }
/// <summary>Gets the number of documents applied into the local node cache.</summary>
public int AppliedCount => AppliedDocuments.Count;
/// <summary>Gets the source shard log sequence reported by the snapshot, when present.</summary>
public long ShardLogSequence => Snapshot.ShardLogSequence ?? 0L;
}
/// <summary>
/// Typed snapshot surface for one remote CultNet endpoint.
/// </summary>
@ -132,6 +158,24 @@ namespace GameCult.Mesh
return CultMesh.FetchSnapshotAsync(Endpoint, CreateRequest(schemaIds, recordKeys));
}
/// <summary>Fetches one snapshot and applies it into the node's cache.</summary>
public async Task<CultMeshSnapshotSyncResult> SyncSnapshotAsync(
CultMeshNode node,
IReadOnlyList<string>? schemaIds = null,
IReadOnlyList<string>? recordKeys = null,
bool flush = false)
{
if (node == null) throw new ArgumentNullException(nameof(node));
var snapshot = await FetchSnapshotAsync(schemaIds, recordKeys).ConfigureAwait(false);
var applied = await node.Database.Documents.ApplyRawSnapshotResponseAsync(node.Cache, snapshot)
.ConfigureAwait(false);
if (flush)
await node.FlushAsync().ConfigureAwait(false);
return new CultMeshSnapshotSyncResult(snapshot, applied);
}
/// <summary>Fetches and decodes documents assignable to the requested type or matching its schema.</summary>
public Task<IReadOnlyList<TDocument>> FetchDocumentsAsync<TDocument>(
IReadOnlyList<string>? recordKeys = null,
@ -145,6 +189,26 @@ namespace GameCult.Mesh
DocumentRegistry);
}
/// <summary>Fetches typed documents and syncs their raw snapshot into the node's cache.</summary>
public async Task<IReadOnlyList<TDocument>> SyncDocumentsAsync<TDocument>(
CultMeshNode node,
IReadOnlyList<string>? recordKeys = null,
IReadOnlyList<string>? schemaIds = null,
bool flush = false)
where TDocument : class
{
if (node == null) throw new ArgumentNullException(nameof(node));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var result = await SyncSnapshotAsync(
node,
schemaIds ?? new[] { descriptor.SchemaId },
recordKeys,
flush)
.ConfigureAwait(false);
return CultMesh.DecodeSnapshotDocuments<TDocument>(result.Snapshot, DocumentRegistry);
}
/// <summary>Fetches one typed document by record key.</summary>
public async Task<TDocument> FetchDocumentAsync<TDocument>(string recordKey)
where TDocument : class
@ -156,6 +220,21 @@ namespace GameCult.Mesh
$"CultNet snapshot endpoint '{Endpoint}' did not return {typeof(TDocument).FullName} record '{recordKey}'.");
}
/// <summary>Fetches one typed document by record key and syncs it into the node's cache.</summary>
public async Task<TDocument> SyncDocumentAsync<TDocument>(
CultMeshNode node,
string recordKey,
bool flush = false)
where TDocument : class
{
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
var documents = await SyncDocumentsAsync<TDocument>(node, new[] { recordKey }, flush: flush)
.ConfigureAwait(false);
return documents.FirstOrDefault()
?? throw new InvalidOperationException(
$"CultNet snapshot endpoint '{Endpoint}' did not return {typeof(TDocument).FullName} record '{recordKey}'.");
}
/// <summary>Creates a typed document handle over one remote endpoint record.</summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
string recordKey,
@ -402,7 +481,7 @@ namespace GameCult.Mesh
return filtered.Length == 0 ? null : filtered;
}
private static IReadOnlyList<TDocument> DecodeSnapshotDocuments<TDocument>(
internal static IReadOnlyList<TDocument> DecodeSnapshotDocuments<TDocument>(
CultNetSnapshotResponseRawMessage snapshot,
CultNetDocumentRegistry registry)
where TDocument : class

View file

@ -1135,15 +1135,33 @@ public sealed class CultMeshStreamingTests
});
var fetchedAlias = await surface.FetchDocumentAsync<MeshNoteAliasDocument>(key.Value);
using var node = await CultMesh.CreateNodeAsync(
Path.Combine(Path.GetTempPath(), $"cultmesh-snapshot-endpoint-{Guid.NewGuid():N}.ccmp"),
new CultMeshNodeOptions
{
StartServer = false,
CacheOptions = new CultCacheOpenOptions
{
Registry = sourceCache.Registry,
PullOnOpen = false
},
DatabaseOptions = new CultNetDatabaseOptions
{
DocumentRegistry = registry
}
});
var syncedAlias = await surface.SyncDocumentAsync<MeshNoteAliasDocument>(node, key.Value);
var aliasHandle = surface.Document<MeshNoteAliasDocument>(key.Value);
var aliasLatest = await aliasHandle.LatestAsync();
var catalog = surface.Documents(surface.Document<MeshNoteDocument>(key.Value));
var catalogAlias = await catalog.LatestAsync<MeshNoteAliasDocument>();
fetchedAlias.Text.Should().Be("snapshot-endpoint");
syncedAlias.Text.Should().Be("snapshot-endpoint");
node.Cache.Get<MeshNoteDocument>(key)!.Revision.Should().Be(19);
aliasLatest.Revision.Should().Be(19);
catalogAlias.Text.Should().Be("snapshot-endpoint");
requests.Should().HaveCount(3);
requests.Should().HaveCount(4);
requests.Should().OnlyContain(request =>
request.SchemaIds!.SequenceEqual(new[]
{