CultLib/src/GameCult.Mesh/CultMesh.cs
Meta Vens 790385364c CultNet selection cut1 fix batch: R-A/B/C/D/E/F/G/H/J/M (C#)
Wires v1 into both C# servers (R-A): CultNetDatabaseServer registers
CultNetSnapshotRequestV1Message/CultNetDatabaseSubscribeV1Message
listeners and Server.cs's pre-verification allowlist gains them;
CultNetDatabaseSubscriptionServer's subscribe path is generalised to a
shared SubscriptionRequest so v0 (lowered) and v1 selections run the
same Subscribe/Watch/Reconcile core, with D6's rule wired in: a
hop-bearing selection runs the full reconcile diff on every change
instead of the single-row fast path. CultNetDatabase.CurrentAsOf gives
both servers a real snapshot watermark.

Fixes the confirmed parity and correctness defects from Soul's
whole-cut pass:
- R-B: cited/cites edges are anchored to the row the direction actually
  reports against (the citer under cites, the cited row under cited),
  not filtered by "From is on the page" regardless of direction; order
  is page-row order then (from, role, to) in code-point order.
- R-C: a new CultNetCodePointComparer replaces every StringComparer.Ordinal/
  string.CompareOrdinal in the row tiebreak, cursor position and edge
  order, so an astral key sorts the same way here as it does under
  Rust's UTF-8 byte comparison.
- R-D: CultCache renders a float/double's exact decimal expansion from
  its IEEE-754 bits via BigInteger (RenderExactFloat/RenderExactDouble),
  never .NET's shortest round-trip form, which rounded a tie the
  opposite way from Rust at some values. The now-dead ExpandScientificNotation
  is deleted.
- R-E: MatchesCitation compares a cites target through
  CultNetSchemaAliasMatching instead of an exact SchemaId string
  compare; ToEdge and ToRawRecord now emit the same wire schema id
  (WireSchemaId) for the same row, where ToEdge previously used the
  descriptor's own id and ToRawRecord the binding's override.
- R-F: CultNetSelectionEvaluator.Select validates first, unconditionally
  - the validate:false/true flag on SelectPage/CreateRawSnapshotResponse
  is gone, so no public evaluation path can skip the door. Mesh's
  EnsureV0Compatible runs the new CultNetSelectionValidation.ValidateShape
  (the declaration-independent half of the door) before its own v0
  transport refusals, so Keys=[]/[""] is refused there too.
- R-G: Evaluation.TotalMatched is the selection's whole matching count;
  SelectPage's Matched field now reports it instead of the page's row
  count. CreateRawSnapshotResponse and CreateShardSnapshotResponse
  (v0/shard full walks) call the new CultNetDocumentRegistry.SelectAll,
  which evaluates once via CultNetSelectionEvaluator.EvaluateAll and
  projects every matched row, instead of looping SelectPage per 200-row
  page and re-filtering/re-sorting the whole row set each time.
- R-H: the cursor digest is length-prefixed for every string and list
  (AppendString/AppendList) instead of delimiter-joined, closing the
  collision ["a|b"] == ["a","b"] and ["a,b"] == ["a","b"].
- R-J: any_of on a numeric alias compares TryGetIndexNumber's canonical
  rendering, not the string getter's culture-dependent ToString().
- R-M: WithBindingSchemaAlias is deleted; both servers' single-change
  fast path calls CultNetDocumentRegistry.ExpandSchemaBindingAliases
  instead, the one binding-alias reconciler. The duplicated v0/shard
  lower-and-page loop is gone with SelectAll above.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-22 09:39:04 +02:00

3515 lines
154 KiB
C#

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
using GameCult.Networking;
using R3;
namespace GameCult.Mesh
{
/// <summary>
/// Options for creating a local CultMesh node.
/// </summary>
public sealed class CultMeshNodeOptions
{
/// <summary>
/// Gets or sets the cache-open options used for the underlying CultCache.
/// </summary>
public CultCacheOpenOptions? CacheOptions { get; set; }
/// <summary>
/// Gets or sets the server security configuration. When omitted, development defaults are used.
/// </summary>
public ServerSecurityOptions? Security { get; set; }
/// <summary>
/// Gets or sets whether the server should start immediately after creation.
/// </summary>
public bool StartServer { get; set; } = true;
/// <summary>
/// Gets or sets the mesh database sharding options used for the hosted CultCache.
/// </summary>
public CultNetDatabaseOptions? DatabaseOptions { get; set; }
/// <summary>
/// Gets or sets whether CultMesh should attach a file-backed authoritative shard-log store when none is supplied.
/// </summary>
public bool EnableDurableShardLogs { get; set; }
/// <summary>
/// Gets or sets the directory used for file-backed authoritative shard logs.
/// </summary>
public string? ShardLogPath { get; set; }
/// <summary>
/// Gets or sets the database server bridge options used for shard routing and forwarding.
/// </summary>
public CultNetDatabaseServerOptions? DatabaseServerOptions { get; set; }
/// <summary>
/// Gets or sets an optional callback used to customize the server before start.
/// </summary>
public Action<Server>? ConfigureServer { get; set; }
internal CultNetHostOptions ToCultNetOptions(string cachePath)
{
return new CultNetHostOptions
{
CacheOptions = CacheOptions,
Security = Security,
StartServer = StartServer,
DatabaseOptions = CreateDatabaseOptions(cachePath),
DatabaseServerOptions = DatabaseServerOptions,
ConfigureServer = ConfigureServer
};
}
private CultNetDatabaseOptions? CreateDatabaseOptions(string cachePath)
{
if (!EnableDurableShardLogs)
{
return DatabaseOptions;
}
var source = DatabaseOptions ?? new CultNetDatabaseOptions();
if (source.MutationLogStore != null)
{
return source;
}
return new CultNetDatabaseOptions
{
RuntimeId = source.RuntimeId,
Shards = source.Shards,
ClientAuthorityScopes = source.ClientAuthorityScopes,
DocumentRegistry = source.DocumentRegistry,
MutationLogStore = new CultNetFileShardMutationLogStore(ResolveShardLogPath(cachePath))
};
}
private string ResolveShardLogPath(string cachePath)
{
if (!string.IsNullOrWhiteSpace(ShardLogPath))
{
return ShardLogPath!;
}
var fullPath = Path.GetFullPath(cachePath);
var directory = Path.GetDirectoryName(fullPath) ?? Directory.GetCurrentDirectory();
var name = Path.GetFileNameWithoutExtension(fullPath);
return Path.Combine(directory, name + ".cultmesh", "shard-logs");
}
}
/// <summary>
/// Options for opening one typed document from a remote CultNet snapshot endpoint.
/// </summary>
public sealed record CultMeshPeerSnapshotDocumentOptions
{
/// <summary>Gets or sets the semantic document id. Defaults to the record key.</summary>
public string? DocumentId { get; set; }
/// <summary>Gets or sets the source id advertised by diagnostics. Defaults to the record key.</summary>
public string? SourceId { get; set; }
/// <summary>Gets or sets the route hint for the resulting handle.</summary>
public CultMeshRouteHint? RouteHint { get; set; }
/// <summary>Gets or sets the response timeout for each snapshot request.</summary>
public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>Gets or sets the connection timeout for endpoint-created clients.</summary>
public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>Gets or sets the polling interval for watch fallback.</summary>
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
/// <summary>Gets or sets the message id prefix used for snapshot requests.</summary>
public string? MessageIdPrefix { get; set; }
/// <summary>Gets or sets the selection to request. Absent means the document type's own schema.</summary>
public CultNetSelection? Selection { get; set; }
/// <summary>Gets or sets the target shard id, when the endpoint is shard-aware.</summary>
public string? ShardId { get; set; }
/// <summary>Gets or sets the target shard epoch, when the endpoint is shard-aware.</summary>
public long? ShardEpoch { get; set; }
/// <summary>Gets or sets client security options for endpoint-created clients.</summary>
public ClientSecurityOptions? Security { get; set; }
/// <summary>Gets or sets a callback used to configure endpoint-created LiteNetLib clients.</summary>
public Action<Client>? ConfigureClient { get; set; }
/// <summary>Gets or sets the runtime id used by endpoint-created RUDP clients.</summary>
public string? RudpRuntimeId { get; set; }
/// <summary>Gets or sets the connection id used by endpoint-created RUDP clients.</summary>
public uint RudpConnectionId { get; set; } = 0x43554c54;
/// <summary>Gets or sets the connect payload used by endpoint-created RUDP clients.</summary>
public string RudpConnectPayload { get; set; } = "cultnet-schema-rudp";
/// <summary>Gets or sets the maximum fragment size used by endpoint-created RUDP clients.</summary>
public int RudpMaxFragmentBytes { get; set; } = 1024;
/// <summary>Gets or sets the resend delay used by endpoint-created RUDP clients.</summary>
public long RudpResendDelayMs { get; set; } = 25;
}
/// <summary>
/// Options for opening one typed document from a CultCache backing-store publication.
/// </summary>
public sealed class CultMeshStoreDocumentOptions
{
/// <summary>Gets or sets the semantic document id. Defaults to the record key.</summary>
public string? DocumentId { get; set; }
/// <summary>Gets or sets the source id advertised by diagnostics. Defaults to the record key.</summary>
public string? SourceId { get; set; }
/// <summary>Gets or sets the route hint for the resulting handle.</summary>
public CultMeshRouteHint? RouteHint { get; set; }
/// <summary>Gets or sets the polling interval for watch fallback.</summary>
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
/// <summary>Gets or sets the document registry used to resolve persisted schema aliases.</summary>
public CultDocumentRegistry? Registry { get; set; }
}
/// <summary>
/// Describes where a typed document publication should be read from.
/// </summary>
public abstract class CultMeshDocumentPublicationSource
{
private CultMeshDocumentPublicationSource()
{
}
/// <summary>Creates a publication source from a CultCache backing store.</summary>
public static CultMeshDocumentPublicationSource Store(CacheBackingStore store)
{
return new StoreSource(store);
}
/// <summary>Creates a publication source from a single-file MessagePack CultCache publication.</summary>
public static CultMeshDocumentPublicationSource SingleFile(string path)
{
return new SingleFileSource(path);
}
/// <summary>Creates a publication source from a remote CultNet snapshot response provider.</summary>
public static CultMeshDocumentPublicationSource PeerSnapshot(Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> snapshot)
{
return new PeerSnapshotSource(snapshot);
}
/// <summary>Creates a publication source from a remote CultNet schema endpoint.</summary>
public static CultMeshDocumentPublicationSource PeerSnapshot(string endpoint)
{
return new PeerSnapshotEndpointSource(endpoint);
}
/// <summary>Creates a publication source from a remote CultNet schema client factory.</summary>
public static CultMeshDocumentPublicationSource PeerSnapshot(Func<ICultNetSchemaClient> createClient, string endpoint)
{
return new PeerSnapshotClientSource(createClient, endpoint);
}
internal sealed class StoreSource : CultMeshDocumentPublicationSource
{
public StoreSource(CacheBackingStore store)
{
BackingStore = store ?? throw new ArgumentNullException(nameof(store));
}
public CacheBackingStore BackingStore { get; }
}
internal sealed class SingleFileSource : CultMeshDocumentPublicationSource
{
public SingleFileSource(string path)
{
Path = string.IsNullOrWhiteSpace(path)
? throw new ArgumentException("Value must be non-empty.", nameof(path))
: path;
}
public string Path { get; }
}
internal sealed class PeerSnapshotSource : CultMeshDocumentPublicationSource
{
public PeerSnapshotSource(Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> snapshot)
{
Snapshot = snapshot ?? throw new ArgumentNullException(nameof(snapshot));
}
public Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> Snapshot { get; }
}
internal sealed class PeerSnapshotEndpointSource : CultMeshDocumentPublicationSource
{
public PeerSnapshotEndpointSource(string endpoint)
{
Endpoint = string.IsNullOrWhiteSpace(endpoint)
? throw new ArgumentException("Value must be non-empty.", nameof(endpoint))
: endpoint;
}
public string Endpoint { get; }
}
internal sealed class PeerSnapshotClientSource : CultMeshDocumentPublicationSource
{
public PeerSnapshotClientSource(Func<ICultNetSchemaClient> createClient, string endpoint)
{
CreateClient = createClient ?? throw new ArgumentNullException(nameof(createClient));
Endpoint = string.IsNullOrWhiteSpace(endpoint)
? throw new ArgumentException("Value must be non-empty.", nameof(endpoint))
: endpoint;
}
public Func<ICultNetSchemaClient> CreateClient { get; }
public string Endpoint { get; }
}
}
/// <summary>
/// Describes one typed document to bind from a CultMesh publication source.
/// </summary>
public interface ICultMeshPublicationDocumentBinding
{
/// <summary>Gets the publication source override for this document, when it differs from the catalog source.</summary>
CultMeshDocumentPublicationSource? Source { get; }
/// <summary>Binds this document request to a publication-backed handle.</summary>
ICultMeshDocumentHandle Bind(
CultMeshDocumentPublicationSource source,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions,
CultMeshPeerSnapshotDocumentOptions? peerOptions);
/// <summary>Reads this publication document and hydrates it into a local node.</summary>
Task<ICultMeshDocumentHandle> SyncAsync(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions,
CultMeshPeerSnapshotDocumentOptions? peerOptions);
}
/// <summary>
/// Describes one typed document to bind from a CultMesh publication source.
/// </summary>
public sealed class CultMeshPublicationDocumentBinding<TDocument> : ICultMeshPublicationDocumentBinding
where TDocument : class
{
internal CultMeshPublicationDocumentBinding(
CultRecordKey key,
string? documentId,
string? sourceId,
CultMeshDocumentPublicationSource? source)
{
Key = key;
DocumentId = documentId;
SourceId = sourceId;
Source = source;
}
/// <summary>Gets the record key to read.</summary>
public CultRecordKey Key { get; }
/// <summary>Gets the semantic document id. Defaults to the record key.</summary>
public string? DocumentId { get; }
/// <summary>Gets the source id advertised by diagnostics. Defaults to the document id or record key.</summary>
public string? SourceId { get; }
/// <inheritdoc />
public CultMeshDocumentPublicationSource? Source { get; }
/// <inheritdoc />
public ICultMeshDocumentHandle Bind(
CultMeshDocumentPublicationSource source,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions,
CultMeshPeerSnapshotDocumentOptions? peerOptions)
{
return CultMesh.DocumentFromPublication<TDocument>(
Source ?? source,
Key,
context,
CultMesh.WithPublicationBindingOptions(storeOptions, Key, DocumentId, SourceId),
CultMesh.WithPublicationBindingOptions(peerOptions, Key, DocumentId, SourceId));
}
/// <inheritdoc />
public async Task<ICultMeshDocumentHandle> SyncAsync(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions,
CultMeshPeerSnapshotDocumentOptions? peerOptions)
{
if (node == null) throw new ArgumentNullException(nameof(node));
if (context == null) throw new ArgumentNullException(nameof(context));
var publication = CultMesh.DocumentFromPublication<TDocument>(
Source ?? source,
Key,
context,
CultMesh.WithPublicationBindingOptions(storeOptions, Key, DocumentId, SourceId),
CultMesh.WithPublicationBindingOptions(peerOptions, Key, DocumentId, SourceId));
var local = CultMesh.Document<TDocument>(
node,
Key,
context,
string.IsNullOrWhiteSpace(DocumentId) ? Key.Value : DocumentId);
await local.ReplaceAsync(await publication.LatestAsync().ConfigureAwait(false)).ConfigureAwait(false);
return local;
}
}
/// <summary>
/// A locally hosted CultMesh node over CultCache, CultNet transport, and the mesh database facade.
/// </summary>
public sealed class CultMeshNode : IDisposable
{
private readonly CultNetHost _host;
internal CultMeshNode(CultNetHost host)
{
_host = host ?? throw new ArgumentNullException(nameof(host));
}
/// <summary>
/// Gets the durable cache used by the node.
/// </summary>
public CultCache Cache => _host.Cache;
/// <summary>
/// Gets the canonical single-file MessagePack store, when one is attached.
/// </summary>
public SingleFileMessagePackBackingStore? Store => _host.Store;
/// <summary>
/// Gets the underlying CultNet server.
/// </summary>
public Server Server => _host.Server;
/// <summary>
/// Gets the distributed realtime database facade.
/// </summary>
public CultNetDatabase Database => _host.Database;
/// <summary>
/// Gets the schema-v0 server bridge for distributed database messages.
/// </summary>
public CultNetDatabaseServer DatabaseServer => _host.DatabaseServer;
/// <summary>
/// Starts the underlying server.
/// </summary>
public void Start()
{
_host.Start();
}
/// <summary>
/// Flushes the durable cache.
/// </summary>
public Task FlushAsync(bool soft = false)
{
return _host.FlushAsync();
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultRecordKey key,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
return CultMesh.Document<TDocument>(Database, key, context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultRecordKey key,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(key, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
string recordKey,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
return Document<TDocument>(new CultRecordKey(recordKey), context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one distributed record hosted by this node.
/// </summary>
public CultMeshDocumentHandle<TDocument> Document<TDocument>(
string recordKey,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(recordKey, verse.Context, documentId, sources, routeHint);
}
/// <inheritdoc />
public void Dispose()
{
_host.Dispose();
}
}
/// <summary>
/// Parsed CultMesh RUDP endpoint.
/// </summary>
public sealed class CultMeshRudpEndpoint
{
/// <summary>
/// Creates a parsed RUDP endpoint.
/// </summary>
public CultMeshRudpEndpoint(string host, int port, string uri)
{
Host = string.IsNullOrWhiteSpace(host)
? throw new ArgumentException("Host must be non-empty.", nameof(host))
: host;
if (port <= 0 || port > 65535)
{
throw new ArgumentOutOfRangeException(nameof(port), "Port must be between 1 and 65535.");
}
Port = port;
Uri = string.IsNullOrWhiteSpace(uri)
? throw new ArgumentException("Uri must be non-empty.", nameof(uri))
: uri;
}
/// <summary>Gets the endpoint host.</summary>
public string Host { get; }
/// <summary>Gets the endpoint port.</summary>
public int Port { get; }
/// <summary>Gets the normalized rudp:// URI.</summary>
public string Uri { get; }
}
/// <summary>
/// Options for CultMesh-branded RUDP socket helpers.
/// </summary>
public sealed class CultMeshRudpSocketOptions
{
/// <summary>Gets or sets the local bind host.</summary>
public string BindHost { get; set; } = "0.0.0.0";
/// <summary>Gets or sets the local bind port.</summary>
public int BindPort { get; set; }
/// <summary>Gets or sets a caller-owned bound socket.</summary>
public Socket? Socket { get; set; }
/// <summary>Gets or sets the first local packet sequence.</summary>
public uint InitialSequence { get; set; } = 1;
/// <summary>Gets or sets the reliable resend delay in milliseconds.</summary>
public long ResendDelayMs { get; set; } = 250;
/// <summary>Gets or sets the advertised transport id.</summary>
public string TransportId { get; set; } = "rudp";
/// <summary>Gets or sets the advertised maximum payload size.</summary>
public int? MaxPayloadBytes { get; set; }
/// <summary>Gets or sets the maximum RUDP fragment size.</summary>
public int? MaxFragmentBytes { get; set; }
/// <summary>Gets or sets the maximum pending reliable packet count.</summary>
public int? MaxPendingReliablePackets { get; set; }
/// <summary>Gets or sets the advertised reconnect policy.</summary>
public CultNetReconnectPolicy? ReconnectPolicy { get; set; }
}
/// <summary>
/// Options for CultMesh-branded connected RUDP client helpers.
/// </summary>
public sealed class CultMeshRudpClientOptions
{
/// <summary>Gets or sets the socket construction and transport profile options.</summary>
public CultMeshRudpSocketOptions SocketOptions { get; set; } = new();
/// <summary>Gets or sets the handshake payload.</summary>
public byte[] ConnectPayload { get; set; } = Array.Empty<byte>();
/// <summary>Gets or sets the handshake timeout.</summary>
public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(1);
/// <summary>Gets or sets the handshake polling interval.</summary>
public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(5);
}
/// <summary>
/// Public CultMesh entrypoints.
/// </summary>
public static partial class CultMesh
{
/// <summary>
/// Creates a local CultMesh node with a durable CultCache and the canonical MessagePack backing store.
/// </summary>
public static async Task<CultMeshNode> CreateNodeAsync(string cachePath, CultMeshNodeOptions? options = null)
{
var host = await CultNetLocal.CreateHostAsync(
cachePath,
(options ?? new CultMeshNodeOptions()).ToCultNetOptions(cachePath)).ConfigureAwait(false);
return new CultMeshNode(host);
}
/// <summary>
/// Creates and starts a local CultMesh node with development-friendly defaults.
/// </summary>
public static Task<CultMeshNode> StartNodeAsync(string cachePath, CultMeshNodeOptions? options = null)
{
options ??= new CultMeshNodeOptions();
options.StartServer = true;
return CreateNodeAsync(cachePath, options);
}
/// <summary>
/// Creates a local reactive catalog for discovered Verses.
/// </summary>
public static CultMeshVerseCatalog CreateVerseCatalog()
{
return new CultMeshVerseCatalog();
}
/// <summary>
/// Creates a local reactive catalog for discovered peers.
/// </summary>
public static CultMeshPeerCatalog CreatePeerCatalog()
{
return new CultMeshPeerCatalog();
}
/// <summary>
/// Creates a local authority lease catalog.
/// </summary>
public static CultMeshAuthorityLeaseCatalog CreateAuthorityLeaseCatalog()
{
return new CultMeshAuthorityLeaseCatalog();
}
/// <summary>
/// Creates a local stream catalog for zero-copy-oriented audio, video, tensor, and byte frame streams.
/// </summary>
public static CultMeshStreamCatalog CreateStreamCatalog()
{
return new CultMeshStreamCatalog();
}
/// <summary>
/// Packs bytes into content-addressed CDN chunk documents and a versioned artifact manifest.
/// </summary>
public static CultMeshCdnArtifact PackCdnArtifact(
string artifactId,
byte[] payload,
CultMeshCdnPackOptions? options = null)
{
return CultMeshCdn.PackArtifact(artifactId, payload, options);
}
/// <summary>
/// Publishes a packed CDN artifact into a local cache.
/// </summary>
public static Task<CultMeshCdnArtifactManifest> PublishCdnArtifactAsync(
CultCache cache,
CultMeshCdnArtifact artifact)
{
return CultMeshCdn.PublishAsync(cache, artifact);
}
/// <summary>
/// Publishes a packed CDN artifact into a distributed CultNet database.
/// </summary>
public static Task<CultMeshCdnArtifactManifest> PublishCdnArtifactAsync(
CultNetDatabase database,
CultMeshCdnArtifact artifact)
{
return CultMeshCdn.PublishAsync(database, artifact);
}
/// <summary>
/// Reassembles and verifies a CDN artifact from a local cache.
/// </summary>
public static byte[] ReadCdnArtifact(
CultCache cache,
CultMeshCdnArtifactManifest manifest)
{
return CultMeshCdn.ReadArtifact(cache, manifest);
}
/// <summary>
/// Reassembles and verifies a CDN artifact from a distributed CultNet database.
/// </summary>
public static Task<byte[]> ReadCdnArtifactAsync(
CultNetDatabase database,
CultMeshCdnArtifactManifest manifest)
{
return CultMeshCdn.ReadArtifactAsync(database, manifest);
}
/// <summary>
/// Creates a CultNet document registry for CDN artifact manifests and chunks.
/// </summary>
public static CultNetDocumentRegistry CreateCdnDocumentRegistry(CultDocumentRegistry? documents = null)
{
return CultMeshCdn.CreateDocumentRegistry(documents);
}
/// <summary>
/// Computes and stamps the content hash for a portable entity prefab package.
/// </summary>
public static CultMeshEntityPrefabPackage FinalizeEntityPrefabPackage(
CultMeshEntityPrefabPackage package)
{
return CultMeshEntityPrefabs.FinalizePackage(package);
}
/// <summary>
/// Publishes a portable entity prefab package into a local cache.
/// </summary>
public static Task<CultMeshEntityPrefabPackage> PublishEntityPrefabPackageAsync(
CultCache cache,
CultMeshEntityPrefabPackage package)
{
return CultMeshEntityPrefabs.PublishAsync(cache, package);
}
/// <summary>
/// Publishes a portable entity prefab package into a distributed CultNet database.
/// </summary>
public static Task<CultMeshEntityPrefabPackage> PublishEntityPrefabPackageAsync(
CultNetDatabase database,
CultMeshEntityPrefabPackage package)
{
return CultMeshEntityPrefabs.PublishAsync(database, package);
}
/// <summary>
/// Creates a CultNet document registry for portable entity prefab packages.
/// </summary>
public static CultNetDocumentRegistry CreateEntityPrefabDocumentRegistry(CultDocumentRegistry? documents = null)
{
return CultMeshEntityPrefabs.CreateDocumentRegistry(documents);
}
/// <summary>
/// Creates a CultNet document registry for CDN assets and portable entity prefab packages.
/// </summary>
public static CultNetDocumentRegistry CreateAssetPipelineDocumentRegistry(CultDocumentRegistry? documents = null)
{
return CultMeshEntityPrefabs.CreateAssetPipelineDocumentRegistry(documents);
}
/// <summary>
/// Creates a managed Verse context for generated domain sugar.
/// </summary>
public static CultMeshVerse Verse(
string verseId,
string runtimeId,
CultMeshRouteHint? routeHint = null,
IEnumerable<CultMeshAuthorityClaim>? claims = null)
{
return new CultMeshVerse(new CultMeshVerseContext(verseId, runtimeId, routeHint, claims));
}
/// <summary>
/// Creates a managed Verse context for generated domain sugar.
/// </summary>
public static Task<CultMeshVerse> ConnectVerseAsync(
string verseId,
string runtimeId,
CultMeshRouteHint? routeHint = null,
IEnumerable<CultMeshAuthorityClaim>? claims = null)
{
return Task.FromResult(Verse(verseId, runtimeId, routeHint, claims));
}
/// <summary>
/// Starts a fluent typed operation context for one runtime.
/// </summary>
public static CultMeshOperationContextBuilder OperationContextFor(string runtimeId)
{
return new CultMeshOperationContextBuilder(runtimeId);
}
/// <summary>
/// Starts a fluent typed query context for one runtime.
/// </summary>
public static CultMeshQueryContextBuilder QueryContextFor(string runtimeId)
{
return new CultMeshQueryContextBuilder(runtimeId);
}
/// <summary>
/// Binds a typed operation handle to a Verse context.
/// </summary>
public static CultMeshBoundOperationHandle<TRequest, TResponse> BindOperation<TRequest, TResponse>(
CultMeshVerseContext context,
CultMeshOperationHandle<TRequest, TResponse> operation)
{
return new CultMeshBoundOperationHandle<TRequest, TResponse>(context, operation);
}
/// <summary>
/// Binds a typed operation handle to a Verse.
/// </summary>
public static CultMeshBoundOperationHandle<TRequest, TResponse> BindOperation<TRequest, TResponse>(
CultMeshVerse verse,
CultMeshOperationHandle<TRequest, TResponse> operation)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindOperation(verse.Context, operation);
}
/// <summary>
/// Binds a typed query surface to a Verse context.
/// </summary>
public static CultMeshBoundQuerySurface<TParameters, TResult> BindQuery<TParameters, TResult>(
CultMeshVerseContext context,
CultMeshQuerySurface<TParameters, TResult> query)
{
return new CultMeshBoundQuerySurface<TParameters, TResult>(context, query);
}
/// <summary>
/// Binds a typed query surface to a Verse.
/// </summary>
public static CultMeshBoundQuerySurface<TParameters, TResult> BindQuery<TParameters, TResult>(
CultMeshVerse verse,
CultMeshQuerySurface<TParameters, TResult> query)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindQuery(verse.Context, query);
}
/// <summary>
/// Binds a typed live feed to a Verse context.
/// </summary>
public static CultMeshBoundLiveFeed<TParameters, TResult> BindLiveFeed<TParameters, TResult>(
CultMeshVerseContext context,
CultMeshLiveFeed<TParameters, TResult> feed)
{
return new CultMeshBoundLiveFeed<TParameters, TResult>(context, feed);
}
/// <summary>
/// Binds a typed live feed to a Verse.
/// </summary>
public static CultMeshBoundLiveFeed<TParameters, TResult> BindLiveFeed<TParameters, TResult>(
CultMeshVerse verse,
CultMeshLiveFeed<TParameters, TResult> feed)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindLiveFeed(verse.Context, feed);
}
/// <summary>
/// Binds a typed document live feed to a Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> BindDocument<TDocument>(
CultMeshVerseContext context,
CultMeshLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed)
where TDocument : class
{
return new CultMeshDocumentHandle<TDocument>(BindLiveFeed(context, feed));
}
/// <summary>
/// Binds a mutable typed document live feed to a Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> BindDocument<TDocument>(
CultMeshVerseContext context,
CultMeshLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed,
Func<TDocument, Task> replace)
where TDocument : class
{
if (replace == null) throw new ArgumentNullException(nameof(replace));
return new CultMeshDocumentHandle<TDocument>(BindLiveFeed(context, feed), replace);
}
/// <summary>
/// Binds a mutable predicted typed document live feed to a Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> BindDocument<TDocument>(
CultMeshVerseContext context,
CultMeshLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed,
Func<TDocument, Task> replace,
Func<TDocument, Task> submitPrediction)
where TDocument : class
{
if (replace == null) throw new ArgumentNullException(nameof(replace));
if (submitPrediction == null) throw new ArgumentNullException(nameof(submitPrediction));
return new CultMeshDocumentHandle<TDocument>(BindLiveFeed(context, feed), replace, submitPrediction);
}
/// <summary>
/// Binds a typed document live feed to a Verse.
/// </summary>
public static CultMeshDocumentHandle<TDocument> BindDocument<TDocument>(
CultMeshVerse verse,
CultMeshLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindDocument(verse.Context, feed);
}
/// <summary>
/// Binds a mutable typed document live feed to a Verse.
/// </summary>
public static CultMeshDocumentHandle<TDocument> BindDocument<TDocument>(
CultMeshVerse verse,
CultMeshLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed,
Func<TDocument, Task> replace)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindDocument(verse.Context, feed, replace);
}
/// <summary>
/// Binds a mutable predicted typed document live feed to a Verse.
/// </summary>
public static CultMeshDocumentHandle<TDocument> BindDocument<TDocument>(
CultMeshVerse verse,
CultMeshLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed,
Func<TDocument, Task> replace,
Func<TDocument, Task> submitPrediction)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindDocument(verse.Context, feed, replace, submitPrediction);
}
/// <summary>
/// Binds a typed state pointer to a Verse context.
/// </summary>
public static CultMeshBoundStatePointer<TValue> BindStatePointer<TValue>(
CultMeshVerseContext context,
CultMeshStatePointer<TValue> pointer)
{
return new CultMeshBoundStatePointer<TValue>(context, pointer);
}
/// <summary>
/// Binds a typed state pointer to a Verse.
/// </summary>
public static CultMeshBoundStatePointer<TValue> BindStatePointer<TValue>(
CultMeshVerse verse,
CultMeshStatePointer<TValue> pointer)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindStatePointer(verse.Context, pointer);
}
/// <summary>
/// Binds a mutable typed state pointer to a Verse context.
/// </summary>
public static CultMeshBoundMutableStatePointer<TValue> BindMutableStatePointer<TValue>(
CultMeshVerseContext context,
CultMeshMutableStatePointer<TValue> pointer)
{
return new CultMeshBoundMutableStatePointer<TValue>(context, pointer);
}
/// <summary>
/// Binds a mutable typed state pointer to a Verse.
/// </summary>
public static CultMeshBoundMutableStatePointer<TValue> BindMutableStatePointer<TValue>(
CultMeshVerse verse,
CultMeshMutableStatePointer<TValue> pointer)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return BindMutableStatePointer(verse.Context, pointer);
}
/// <summary>
/// Creates a typed state pointer to Verse state.
/// </summary>
public static CultMeshStatePointer<TValue> StatePointer<TValue>(
string pointerId,
Func<Task<TValue?>> resolve,
Func<R3.Observable<TValue>> watch,
CultMeshRouteHint? routeHint = null,
IEnumerable<CultMeshProjectionSource>? sources = null)
{
return new CultMeshStatePointer<TValue>(
pointerId,
resolve,
watch,
routeHint,
sources);
}
/// <summary>
/// Creates a typed state pointer to Verse state that can resolve through a query context.
/// </summary>
public static CultMeshStatePointer<TValue> StatePointer<TValue>(
string pointerId,
Func<CultMeshQueryContext, Task<TValue?>> resolve,
Func<CultMeshQueryContext, R3.Observable<TValue>> watch,
CultMeshRouteHint? routeHint = null,
IEnumerable<CultMeshProjectionSource>? sources = null)
{
return new CultMeshStatePointer<TValue>(
pointerId,
resolve,
watch,
routeHint,
sources);
}
/// <summary>
/// Creates a mutable typed state pointer to Verse state.
/// </summary>
public static CultMeshMutableStatePointer<TValue> MutableStatePointer<TValue>(
string pointerId,
Func<Task<TValue?>> resolve,
Func<R3.Observable<TValue>> watch,
Func<TValue, Task> replace,
CultMeshRouteHint? routeHint = null,
IEnumerable<CultMeshProjectionSource>? sources = null)
{
return new CultMeshMutableStatePointer<TValue>(
pointerId,
resolve,
watch,
replace,
routeHint,
sources);
}
/// <summary>
/// Creates a mutable typed state pointer to Verse state that can operate through a query context.
/// </summary>
public static CultMeshMutableStatePointer<TValue> MutableStatePointer<TValue>(
string pointerId,
Func<CultMeshQueryContext, Task<TValue?>> resolve,
Func<CultMeshQueryContext, R3.Observable<TValue>> watch,
Func<CultMeshQueryContext, TValue, Task> replace,
CultMeshRouteHint? routeHint = null,
IEnumerable<CultMeshProjectionSource>? sources = null)
{
return new CultMeshMutableStatePointer<TValue>(
pointerId,
resolve,
watch,
replace,
routeHint,
sources);
}
/// <summary>
/// Creates a typed projection source descriptor.
/// </summary>
public static CultMeshProjectionSource ProjectionSource(
string sourceId,
string? schemaId = null,
string? description = null)
{
return new CultMeshProjectionSource(sourceId, schemaId, description);
}
/// <summary>
/// Creates a named state-reference resolver for portable UI/tool surfaces.
/// </summary>
public static CultMeshStateRefResolver StateRefResolver(
string resolverId,
Func<string, string?> resolve,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
{
if (resolve == null) throw new ArgumentNullException(nameof(resolve));
return new CultMeshStateRefResolver(
resolverId,
(stateRef, _context) => resolve(stateRef),
sources,
routeHint);
}
/// <summary>
/// Creates a named state-reference resolver for portable UI/tool surfaces.
/// </summary>
public static CultMeshStateRefResolver StateRefResolver(
string resolverId,
Func<string, CultMeshQueryContext, string?> resolve,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
{
return new CultMeshStateRefResolver(resolverId, resolve, sources, routeHint);
}
/// <summary>
/// Describes a state-reference resolver for catalogs and tools.
/// </summary>
public static CultMeshStateRefResolverDiagnostic DescribeStateRefResolver(
CultMeshStateRefResolver resolver)
{
if (resolver == null) throw new ArgumentNullException(nameof(resolver));
return new CultMeshStateRefResolverDiagnostic(
resolver.ResolverId,
resolver.RouteHint,
resolver.Sources);
}
/// <summary>
/// Creates a UI/tool binding descriptor from a component property to a typed state pointer.
/// </summary>
public static CultMeshStateBindingDescriptor StateBinding<TValue>(
string targetProp,
CultMeshStatePointer<TValue> pointer)
{
return CultMeshStateBindingDescriptor.FromPointer(targetProp, pointer);
}
/// <summary>
/// Creates a UI/tool binding descriptor from a component property to a mutable typed state pointer.
/// </summary>
public static CultMeshStateBindingDescriptor StateBinding<TValue>(
string targetProp,
CultMeshMutableStatePointer<TValue> pointer)
{
if (pointer == null) throw new ArgumentNullException(nameof(pointer));
return CultMeshStateBindingDescriptor.FromPointer(targetProp, pointer.AsStatePointer());
}
/// <summary>
/// Creates a UI/tool binding descriptor from a component property to a state pointer id.
/// </summary>
public static CultMeshStateBindingDescriptor StateBinding(
string targetProp,
string pointerId,
string? sourceId = null,
string? schemaId = null,
CultMeshRouteHint? routeHint = null,
string? bindingName = null,
string? documentId = null,
string? fieldPath = null,
string? valueKind = null,
string? accessMode = null,
string? authority = null,
string? writeCommand = null)
{
return new CultMeshStateBindingDescriptor(
targetProp, pointerId, sourceId, schemaId, routeHint,
bindingName, documentId, fieldPath, valueKind, accessMode, authority, writeCommand);
}
/// <summary>
/// Creates transport-friendly fields from a state binding descriptor.
/// </summary>
public static CultMeshStateBindingRecord StateBindingRecord(CultMeshStateBindingDescriptor? binding)
{
return CultMeshStateBindingRecord.FromBinding(binding);
}
/// <summary>
/// Creates transport-friendly state binding fields directly.
/// </summary>
public static CultMeshStateBindingRecord StateBindingRecord(
string? targetProp,
string? pointerId,
string? sourceId = null,
string? schemaId = null,
string? routeKind = null,
string? routeDescription = null,
string? bindingName = null,
string? documentId = null,
string? fieldPath = null,
string? valueKind = null,
string? accessMode = null,
string? authority = null,
string? writeCommand = null)
{
return new CultMeshStateBindingRecord(
targetProp,
pointerId,
sourceId,
schemaId,
routeKind,
routeDescription,
bindingName,
documentId,
fieldPath,
valueKind,
accessMode,
authority,
writeCommand);
}
/// <summary>
/// Creates a UI/tool command binding descriptor from a typed operation handle.
/// </summary>
public static CultMeshOperationBindingDescriptor OperationBinding<TRequest, TResponse>(
CultMeshOperationHandle<TRequest, TResponse> operation,
string? label = null,
string? schemaId = null,
CultMeshRouteHint? routeHint = null)
{
return CultMeshOperationBindingDescriptor.FromOperation(operation, label, schemaId, routeHint);
}
/// <summary>
/// Creates a UI/tool command binding descriptor from a typed operation id.
/// </summary>
public static CultMeshOperationBindingDescriptor OperationBinding(
string operationId,
string? label = null,
string? schemaId = null,
CultMeshRouteHint? routeHint = null)
{
return new CultMeshOperationBindingDescriptor(operationId, label, schemaId, routeHint);
}
/// <summary>
/// Creates transport-friendly fields from an operation binding descriptor.
/// </summary>
public static CultMeshOperationBindingRecord OperationBindingRecord(
CultMeshOperationBindingDescriptor? binding)
{
return CultMeshOperationBindingRecord.FromBinding(binding);
}
/// <summary>
/// Creates transport-friendly operation binding fields directly.
/// </summary>
public static CultMeshOperationBindingRecord OperationBindingRecord(
string? operationId,
string? label = null,
string? schemaId = null,
string? routeKind = null,
string? routeDescription = null)
{
return new CultMeshOperationBindingRecord(
operationId,
label,
schemaId,
routeKind,
routeDescription);
}
/// <summary>
/// Creates a concrete invocation descriptor from an advertised operation binding.
/// </summary>
public static CultMeshOperationInvocationDescriptor OperationInvocation(
CultMeshOperationBindingDescriptor binding,
string? idempotencyKey = null)
{
return CultMeshOperationInvocationDescriptor.FromBinding(binding, idempotencyKey);
}
/// <summary>
/// Creates a concrete invocation descriptor from a typed operation handle.
/// </summary>
public static CultMeshOperationInvocationDescriptor OperationInvocation<TRequest, TResponse>(
CultMeshOperationHandle<TRequest, TResponse> operation,
string? schemaId = null,
CultMeshRouteHint? routeHint = null,
string? idempotencyKey = null)
{
return CultMeshOperationInvocationDescriptor.FromOperation(
operation,
schemaId,
routeHint,
idempotencyKey);
}
/// <summary>
/// Creates a concrete invocation descriptor from a typed operation id.
/// </summary>
public static CultMeshOperationInvocationDescriptor OperationInvocation(
string operationId,
string? schemaId = null,
CultMeshRouteHint? routeHint = null,
string? idempotencyKey = null)
{
return new CultMeshOperationInvocationDescriptor(operationId, schemaId, routeHint, idempotencyKey);
}
/// <summary>
/// Creates transport-friendly fields from a route hint.
/// </summary>
public static CultMeshRouteRecord RouteRecord(CultMeshRouteHint? routeHint)
{
return CultMeshRouteRecord.FromRoute(routeHint);
}
/// <summary>
/// Creates transport-friendly route fields directly.
/// </summary>
public static CultMeshRouteRecord RouteRecord(string? kind, string? description = null)
{
return new CultMeshRouteRecord(kind, description);
}
/// <summary>
/// Creates transport-friendly fields from a typed operation invocation.
/// </summary>
public static CultMeshOperationInvocationRecord OperationInvocationRecord(
CultMeshOperationInvocationDescriptor? invocation,
string? fallbackOperationId = null,
string? fallbackSchemaId = null,
CultMeshRouteHint? fallbackRouteHint = null,
string? fallbackIdempotencyKey = null)
{
return CultMeshOperationInvocationRecord.FromInvocation(
invocation,
fallbackOperationId,
fallbackSchemaId,
fallbackRouteHint,
fallbackIdempotencyKey);
}
/// <summary>
/// Creates transport-friendly operation invocation fields directly.
/// </summary>
public static CultMeshOperationInvocationRecord OperationInvocationRecord(
string? operationId,
string? schemaId = null,
string? routeKind = null,
string? routeDescription = null,
string? idempotencyKey = null)
{
return new CultMeshOperationInvocationRecord(
operationId,
schemaId,
routeKind,
routeDescription,
idempotencyKey);
}
/// <summary>
/// Creates an empty shared operation payload.
/// </summary>
public static CultMeshOperationPayload OperationPayload()
{
return CultMeshOperationPayload.Empty;
}
/// <summary>
/// Creates a shared operation payload from string-compatible fields.
/// </summary>
public static CultMeshOperationPayload OperationPayload(
IEnumerable<KeyValuePair<string, string>>? fields)
{
return CultMeshOperationPayload.FromStrings(fields);
}
/// <summary>
/// Creates a shared operation payload from key/value pairs.
/// </summary>
public static CultMeshOperationPayload OperationPayload(params (string Key, string Value)[] fields)
{
var pairs = fields?.Select(field => new KeyValuePair<string, string>(field.Key, field.Value));
return CultMeshOperationPayload.FromStrings(pairs);
}
/// <summary>
/// Creates a reusable projection recipe over typed source state.
/// </summary>
public static CultMeshProjectionRecipe<TParameters, TResult> ProjectionRecipe<TParameters, TResult>(
string projectionId,
IEnumerable<CultMeshProjectionSource> sources,
Func<TParameters, CultMeshQueryContext, Task<TResult>> project,
CultMeshRouteHint? routeHint = null,
Func<TParameters, CultMeshQueryContext, R3.Observable<TResult>>? watch = null)
{
return new CultMeshProjectionRecipe<TParameters, TResult>(
projectionId,
sources,
project,
routeHint,
watch);
}
/// <summary>
/// Creates a typed live feed surface for coherent client snapshots.
/// </summary>
public static CultMeshLiveFeed<TParameters, TResult> LiveFeed<TParameters, TResult>(
string feedId,
Func<TParameters, CultMeshQueryContext, Task<TResult>> snapshot,
Func<TParameters, CultMeshQueryContext, R3.Observable<TResult>>? watch = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
{
return new CultMeshLiveFeed<TParameters, TResult>(
feedId,
snapshot,
watch,
sources,
routeHint);
}
/// <summary>
/// Creates a typed document handle from snapshot/watch delegates and binds it to a Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
string documentId,
CultMeshVerseContext context,
Func<CultMeshQueryContext, Task<TDocument>> latest,
Func<CultMeshQueryContext, R3.Observable<TDocument>> watch,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (latest == null) throw new ArgumentNullException(nameof(latest));
if (watch == null) throw new ArgumentNullException(nameof(watch));
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
documentId,
(_parameters, queryContext) => latest(queryContext),
(_parameters, queryContext) => watch(queryContext),
sources,
routeHint);
return BindDocument(context, feed);
}
/// <summary>
/// Creates a mutable typed document handle from snapshot/watch/replace delegates and binds it to a Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
string documentId,
CultMeshVerseContext context,
Func<CultMeshQueryContext, Task<TDocument>> latest,
Func<CultMeshQueryContext, R3.Observable<TDocument>> watch,
Func<TDocument, Task> replace,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (replace == null) throw new ArgumentNullException(nameof(replace));
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
documentId,
(_parameters, queryContext) => latest(queryContext),
(_parameters, queryContext) => watch(queryContext),
sources,
routeHint);
return BindDocument(context, feed, replace);
}
/// <summary>
/// Creates a mutable predicted typed document handle from snapshot/watch/replace/prediction delegates and binds it to a Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
string documentId,
CultMeshVerseContext context,
Func<CultMeshQueryContext, Task<TDocument>> latest,
Func<CultMeshQueryContext, R3.Observable<TDocument>> watch,
Func<TDocument, Task> replace,
Func<TDocument, Task> submitPrediction,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (replace == null) throw new ArgumentNullException(nameof(replace));
if (submitPrediction == null) throw new ArgumentNullException(nameof(submitPrediction));
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
documentId,
(_parameters, queryContext) => latest(queryContext),
(_parameters, queryContext) => watch(queryContext),
sources,
routeHint);
return BindDocument(context, feed, replace, submitPrediction);
}
/// <summary>
/// Creates a typed document handle from snapshot/watch delegates and binds it to a Verse.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
string documentId,
CultMeshVerse verse,
Func<CultMeshQueryContext, Task<TDocument>> latest,
Func<CultMeshQueryContext, R3.Observable<TDocument>> watch,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document(documentId, verse.Context, latest, watch, sources, routeHint);
}
/// <summary>
/// Creates a mutable typed document handle from snapshot/watch/replace delegates and binds it to a Verse.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
string documentId,
CultMeshVerse verse,
Func<CultMeshQueryContext, Task<TDocument>> latest,
Func<CultMeshQueryContext, R3.Observable<TDocument>> watch,
Func<TDocument, Task> replace,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document(documentId, verse.Context, latest, watch, replace, sources, routeHint);
}
/// <summary>
/// Creates a mutable predicted typed document handle from snapshot/watch/replace/prediction delegates and binds it to a Verse.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
string documentId,
CultMeshVerse verse,
Func<CultMeshQueryContext, Task<TDocument>> latest,
Func<CultMeshQueryContext, R3.Observable<TDocument>> watch,
Func<TDocument, Task> replace,
Func<TDocument, Task> submitPrediction,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document(documentId, verse.Context, latest, watch, replace, submitPrediction, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle directly over one CultCache record.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultCache cache,
CultRecordKey key,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (cache == null) throw new ArgumentNullException(nameof(cache));
if (context == null) throw new ArgumentNullException(nameof(context));
var descriptor = cache.Registry.GetRequired<TDocument>();
var sourceList = sources?.ToArray()
?? new[] { ProjectionSource(key.Value, descriptor.SchemaId, "CultCache record") };
var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.InProcess, "CultCache document");
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
ResolveDocumentId(documentId, key),
(_parameters, _context) => Task.FromResult(ReadRequired<TDocument>(cache, key)),
(_parameters, _context) => cache.WatchRecord<TDocument>(key)
.Where(change => change.Document != null)
.Select(change => change.Document!),
sourceList,
route);
return new CultMeshDocumentHandle<TDocument>(
BindLiveFeed(context, feed),
async value => await cache.UpsertAsync(value, new CultRecordHandle<TDocument>(key)).ConfigureAwait(false),
() => cache.WatchRecord<TDocument>(key),
() =>
{
var (document, sequence) = cache.GetWithSequence(key);
return (RequireDocument<TDocument>(document, key), sequence);
});
}
/// <summary>
/// Creates a typed document handle directly over one CultCache record.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultCache cache,
CultRecordKey key,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(cache, key, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle directly over one CultCache record with a local Verse context.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultCache cache,
CultRecordKey key,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
return Document<TDocument>(
cache,
key,
Verse("local", "local", routeHint).Context,
documentId,
sources,
routeHint);
}
/// <summary>
/// Creates a read-only typed document handle over one CultCache backing-store publication.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromStore<TDocument>(
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<TDocument>();
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<CultMeshQueryContext, Task<TDocument>> latest = async _ =>
{
await cache.PullAllBackingStoresAsync().ConfigureAwait(false);
return ReadRequired<TDocument>(cache, key);
};
var watch = PollingQueryWatcher<CultMeshDocumentQueryParameters, TDocument>(
async (_parameters, queryContext) => await latest(queryContext).ConfigureAwait(false),
new CultMeshPollingWatchOptions<TDocument>(resolvedOptions.PollInterval));
return Document<TDocument>(
documentId,
context,
latest,
queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext),
sources,
route);
}
/// <summary>
/// Creates a read-only typed document handle over one CultCache backing-store publication.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromStore<TDocument>(
CacheBackingStore store,
CultRecordKey key,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromStore<TDocument>(store, key, verse.Context, options);
}
/// <summary>
/// Creates a read-only typed document handle over one single-file MessagePack CultCache publication.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromSingleFile<TDocument>(
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<TDocument>(
new SingleFileMessagePackBackingStore(path),
key,
context,
fileOptions);
}
/// <summary>
/// Creates a read-only typed document handle over one single-file MessagePack CultCache publication.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromSingleFile<TDocument>(
string path,
CultRecordKey key,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromSingleFile<TDocument>(path, key, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle from a configured publication source.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPublication<TDocument>(
CultMeshDocumentPublicationSource source,
CultRecordKey key,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null)
where TDocument : class
{
if (source == null) throw new ArgumentNullException(nameof(source));
if (context == null) throw new ArgumentNullException(nameof(context));
switch (source)
{
case CultMeshDocumentPublicationSource.StoreSource store:
return DocumentFromStore<TDocument>(store.BackingStore, key, context, storeOptions);
case CultMeshDocumentPublicationSource.SingleFileSource file:
return DocumentFromSingleFile<TDocument>(file.Path, key, context, storeOptions);
case CultMeshDocumentPublicationSource.PeerSnapshotSource snapshot:
return DocumentFromPeerSnapshot<TDocument>(snapshot.Snapshot, key.Value, context, peerOptions);
case CultMeshDocumentPublicationSource.PeerSnapshotEndpointSource endpoint:
return DocumentFromPeerSnapshot<TDocument>(endpoint.Endpoint, key.Value, context, peerOptions);
case CultMeshDocumentPublicationSource.PeerSnapshotClientSource client:
return DocumentFromPeerSnapshot<TDocument>(client.CreateClient, client.Endpoint, key.Value, context, peerOptions);
default:
throw new NotSupportedException(
$"Unsupported CultMesh document publication source '{source.GetType().FullName}'.");
}
}
/// <summary>
/// Creates a typed document handle from a configured publication source.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPublication<TDocument>(
CultMeshDocumentPublicationSource source,
CultRecordKey key,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPublication<TDocument>(source, key, verse.Context, storeOptions, peerOptions);
}
/// <summary>
/// Describes one typed document to bind from a CultMesh publication source.
/// </summary>
public static CultMeshPublicationDocumentBinding<TDocument> PublicationDocument<TDocument>(
CultRecordKey key,
string? documentId = null,
string? sourceId = null,
CultMeshDocumentPublicationSource? source = null)
where TDocument : class
{
return new CultMeshPublicationDocumentBinding<TDocument>(key, documentId, sourceId, source);
}
/// <summary>
/// Describes one typed document to bind from a CultMesh publication source.
/// </summary>
public static CultMeshPublicationDocumentBinding<TDocument> PublicationDocument<TDocument>(
string recordKey,
string? documentId = null,
string? sourceId = null,
CultMeshDocumentPublicationSource? source = null)
where TDocument : class
{
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
return PublicationDocument<TDocument>(new CultRecordKey(recordKey), documentId, sourceId, source);
}
/// <summary>
/// Creates a schema-aware document catalog from one or more configured publication bindings.
/// </summary>
public static CultMeshDocumentCatalog DocumentsFromPublication(
CultMeshDocumentPublicationSource source,
IEnumerable<ICultMeshPublicationDocumentBinding> bindings,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null)
{
if (source == null) throw new ArgumentNullException(nameof(source));
if (bindings == null) throw new ArgumentNullException(nameof(bindings));
if (context == null) throw new ArgumentNullException(nameof(context));
return Documents(bindings.Select(binding =>
{
if (binding == null) throw new ArgumentException("CultMesh publication bindings cannot contain null.", nameof(bindings));
return binding.Bind(source, context, storeOptions, peerOptions);
}).ToArray());
}
/// <summary>
/// Creates a schema-aware document catalog from one or more configured publication bindings.
/// </summary>
public static CultMeshDocumentCatalog DocumentsFromPublication(
CultMeshDocumentPublicationSource source,
IEnumerable<ICultMeshPublicationDocumentBinding> bindings,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentsFromPublication(source, bindings, verse.Context, storeOptions, peerOptions);
}
/// <summary>
/// Reads configured publication documents, hydrates them into a local node, and returns local handles as a catalog.
/// </summary>
public static async Task<CultMeshDocumentCatalog> SyncDocumentsFromPublicationAsync(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
IEnumerable<ICultMeshPublicationDocumentBinding> bindings,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
bool flush = false)
{
if (node == null) throw new ArgumentNullException(nameof(node));
if (source == null) throw new ArgumentNullException(nameof(source));
if (bindings == null) throw new ArgumentNullException(nameof(bindings));
if (context == null) throw new ArgumentNullException(nameof(context));
var handles = await Task.WhenAll(bindings.Select(binding =>
{
if (binding == null) throw new ArgumentException("CultMesh publication bindings cannot contain null.", nameof(bindings));
return binding.SyncAsync(node, source, context, storeOptions, peerOptions);
})).ConfigureAwait(false);
if (flush)
await node.FlushAsync().ConfigureAwait(false);
return Documents(handles);
}
/// <summary>
/// Reads configured publication documents, hydrates them into a local node, and returns local handles as a catalog.
/// </summary>
public static Task<CultMeshDocumentCatalog> SyncDocumentsFromPublicationAsync(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
IEnumerable<ICultMeshPublicationDocumentBinding> bindings,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
bool flush = false)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return SyncDocumentsFromPublicationAsync(
node,
source,
bindings,
verse.Context,
storeOptions,
peerOptions,
flush);
}
/// <summary>
/// Reads one typed document from a configured publication source and hydrates it into a local node.
/// </summary>
public static async Task<TDocument> SyncDocumentFromPublicationAsync<TDocument>(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
CultRecordKey key,
CultMeshVerseContext context,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
bool flush = false)
where TDocument : class
{
if (node == null) throw new ArgumentNullException(nameof(node));
if (source == null) throw new ArgumentNullException(nameof(source));
if (context == null) throw new ArgumentNullException(nameof(context));
var document = await DocumentFromPublication<TDocument>(
source,
key,
context,
storeOptions,
peerOptions)
.LatestAsync()
.ConfigureAwait(false);
await Document<TDocument>(node, key, context)
.ReplaceAsync(document)
.ConfigureAwait(false);
if (flush)
await node.FlushAsync().ConfigureAwait(false);
return document;
}
/// <summary>
/// Reads one typed document from a configured publication source and hydrates it into a local node.
/// </summary>
public static Task<TDocument> SyncDocumentFromPublicationAsync<TDocument>(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
CultRecordKey key,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
bool flush = false)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return SyncDocumentFromPublicationAsync<TDocument>(
node,
source,
key,
verse.Context,
storeOptions,
peerOptions,
flush);
}
/// <summary>
/// Reads one typed document from a configured publication source and hydrates it into a local node.
/// </summary>
public static Task<TDocument> SyncDocumentFromPublicationAsync<TDocument>(
CultMeshNode node,
CultMeshDocumentPublicationSource source,
string recordKey,
CultMeshVerse verse,
CultMeshStoreDocumentOptions? storeOptions = null,
CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
bool flush = false)
where TDocument : class
{
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
return SyncDocumentFromPublicationAsync<TDocument>(
node,
source,
new CultRecordKey(recordKey),
verse,
storeOptions,
peerOptions,
flush);
}
internal static CultMeshStoreDocumentOptions WithPublicationBindingOptions(
CultMeshStoreDocumentOptions? options,
CultRecordKey key,
string? documentId,
string? sourceId)
{
return new CultMeshStoreDocumentOptions
{
DocumentId = string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId,
SourceId = string.IsNullOrWhiteSpace(sourceId)
? (string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId)
: sourceId,
RouteHint = options?.RouteHint,
PollInterval = options?.PollInterval ?? TimeSpan.FromMilliseconds(250),
Registry = options?.Registry
};
}
internal static CultMeshPeerSnapshotDocumentOptions WithPublicationBindingOptions(
CultMeshPeerSnapshotDocumentOptions? options,
CultRecordKey key,
string? documentId,
string? sourceId)
{
var resolved = options is null ? new CultMeshPeerSnapshotDocumentOptions() : options with { };
resolved.DocumentId = string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId;
resolved.SourceId = string.IsNullOrWhiteSpace(sourceId)
? (string.IsNullOrWhiteSpace(documentId) ? key.Value : documentId)
: sourceId;
return resolved;
}
/// <summary>
/// Creates a typed document handle directly over one distributed CultNet database record.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultNetDatabase database,
CultRecordKey key,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (database == null) throw new ArgumentNullException(nameof(database));
if (context == null) throw new ArgumentNullException(nameof(context));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var sourceList = sources?.ToArray()
?? new[] { ProjectionSource(key.Value, descriptor.SchemaId, "CultNet database record") };
var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Automatic, "CultNet database document");
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
ResolveDocumentId(documentId, key),
(_parameters, _context) => ReadDatabaseDocumentRequiredAsync<TDocument>(database, key),
(_parameters, _context) => WatchDatabaseRecordAs<TDocument>(database, key),
sourceList,
route);
Func<TDocument, Task>? replace = database.CanWriteAuthoritatively<TDocument>(key)
? value => PutDatabaseDocumentAsync(database, key, value, predicted: false)
: null;
Func<TDocument, Task>? submitPrediction = database.CanSubmitPrediction<TDocument>(key)
? value => PutDatabaseDocumentAsync(database, key, value, predicted: true)
: null;
return new CultMeshDocumentHandle<TDocument>(
BindLiveFeed(context, feed),
replace,
submitPrediction);
}
/// <summary>
/// Creates a typed document handle directly over one distributed CultNet database record.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultNetDatabase database,
CultRecordKey key,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(database, key, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet snapshot response provider.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> snapshot,
string recordKey,
CultMeshVerseContext context,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (snapshot == null) throw new ArgumentNullException(nameof(snapshot));
if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
if (context == null) throw new ArgumentNullException(nameof(context));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions();
var documentId = string.IsNullOrWhiteSpace(resolvedOptions.DocumentId)
? recordKey
: resolvedOptions.DocumentId!;
var route = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Network, "CultNet snapshot");
var sources = new[]
{
ProjectionSource(
string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? recordKey : resolvedOptions.SourceId!,
descriptor.SchemaId,
"CultNet snapshot")
};
Func<CultMeshQueryContext, Task<TDocument>> latest = async queryContext =>
ReadDocumentFromSnapshotResponse<TDocument>(
await snapshot(queryContext).ConfigureAwait(false),
descriptor.SchemaId,
recordKey);
var watch = PollingQueryWatcher<CultMeshDocumentQueryParameters, TDocument>(
async (_parameters, queryContext) => await latest(queryContext).ConfigureAwait(false),
new CultMeshPollingWatchOptions<TDocument>(resolvedOptions.PollInterval));
return Document<TDocument>(
documentId,
context,
latest,
queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext),
sources,
route);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet snapshot response provider.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<CultMeshQueryContext, Task<CultNetSnapshotResponseRawMessage>> snapshot,
string recordKey,
CultMeshVerse verse,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPeerSnapshot<TDocument>(snapshot, recordKey, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema endpoint.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
string endpoint,
string recordKey,
CultMeshVerseContext context,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions();
return DocumentFromPeerSnapshot<TDocument>(
queryContext => RequestPeerSnapshotAsync<TDocument>(
() => CultNetSchemaClients.CreateForEndpoint(endpoint, resolvedOptions.Security, resolvedOptions.ConfigureClient),
endpoint,
recordKey,
queryContext,
resolvedOptions),
recordKey,
context,
resolvedOptions);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema endpoint.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
string endpoint,
string recordKey,
CultMeshVerse verse,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPeerSnapshot<TDocument>(endpoint, recordKey, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema client factory.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<ICultNetSchemaClient> createClient,
string endpoint,
string recordKey,
CultMeshVerseContext context,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (createClient == null) throw new ArgumentNullException(nameof(createClient));
if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions();
return DocumentFromPeerSnapshot<TDocument>(
queryContext => RequestPeerSnapshotAsync<TDocument>(
createClient,
endpoint,
recordKey,
queryContext,
resolvedOptions),
recordKey,
context,
resolvedOptions);
}
/// <summary>
/// Creates a typed document handle over one remote CultNet schema client factory.
/// </summary>
public static CultMeshDocumentHandle<TDocument> DocumentFromPeerSnapshot<TDocument>(
Func<ICultNetSchemaClient> createClient,
string endpoint,
string recordKey,
CultMeshVerse verse,
CultMeshPeerSnapshotDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return DocumentFromPeerSnapshot<TDocument>(createClient, endpoint, recordKey, verse.Context, options);
}
/// <summary>
/// Creates a typed document handle directly over one CultMesh node database record.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultMeshNode node,
CultRecordKey key,
CultMeshVerseContext context,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (node == null) throw new ArgumentNullException(nameof(node));
return Document<TDocument>(node.Database, key, context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a typed document handle directly over one CultMesh node database record.
/// </summary>
public static CultMeshDocumentHandle<TDocument> Document<TDocument>(
CultMeshNode node,
CultRecordKey key,
CultMeshVerse verse,
string? documentId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Document<TDocument>(node, key, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a schema-aware catalog over typed document handles.
/// </summary>
public static CultMeshDocumentCatalog Documents(IEnumerable<ICultMeshDocumentHandle> documents)
{
return new CultMeshDocumentCatalog(documents);
}
/// <summary>
/// Creates a schema-aware catalog over typed document handles.
/// </summary>
public static CultMeshDocumentCatalog Documents(params ICultMeshDocumentHandle[] documents)
{
return new CultMeshDocumentCatalog(documents);
}
/// <summary>
/// Creates a schema-aware catalog over typed collection handles.
/// </summary>
public static CultMeshCollectionCatalog Collections(IEnumerable<ICultMeshCollectionHandle> collections)
{
return new CultMeshCollectionCatalog(collections);
}
/// <summary>
/// Creates a schema-aware catalog over typed collection handles.
/// </summary>
public static CultMeshCollectionCatalog Collections(params ICultMeshCollectionHandle[] collections)
{
return new CultMeshCollectionCatalog(collections);
}
/// <summary>
/// Reads two typed document handles and projects them into one caller-defined view.
/// </summary>
public static async Task<TResult> LatestAsync<TFirst, TSecond, TResult>(
CultMeshDocumentHandle<TFirst> first,
CultMeshDocumentHandle<TSecond> second,
Func<TFirst, TSecond, TResult> project)
where TFirst : class
where TSecond : class
{
if (first == null) throw new ArgumentNullException(nameof(first));
if (second == null) throw new ArgumentNullException(nameof(second));
if (project == null) throw new ArgumentNullException(nameof(project));
var firstTask = first.LatestAsync();
var secondTask = second.LatestAsync();
await Task.WhenAll(firstTask, secondTask).ConfigureAwait(false);
return project(
await firstTask.ConfigureAwait(false),
await secondTask.ConfigureAwait(false));
}
/// <summary>
/// Reads three typed document handles and projects them into one caller-defined view.
/// </summary>
public static async Task<TResult> LatestAsync<TFirst, TSecond, TThird, TResult>(
CultMeshDocumentHandle<TFirst> first,
CultMeshDocumentHandle<TSecond> second,
CultMeshDocumentHandle<TThird> third,
Func<TFirst, TSecond, TThird, TResult> project)
where TFirst : class
where TSecond : class
where TThird : class
{
if (first == null) throw new ArgumentNullException(nameof(first));
if (second == null) throw new ArgumentNullException(nameof(second));
if (third == null) throw new ArgumentNullException(nameof(third));
if (project == null) throw new ArgumentNullException(nameof(project));
var firstTask = first.LatestAsync();
var secondTask = second.LatestAsync();
var thirdTask = third.LatestAsync();
await Task.WhenAll(firstTask, secondTask, thirdTask).ConfigureAwait(false);
return project(
await firstTask.ConfigureAwait(false),
await secondTask.ConfigureAwait(false),
await thirdTask.ConfigureAwait(false));
}
/// <summary>
/// Creates a typed collection handle over all local CultCache records assignable to the document type.
/// </summary>
public static CultMeshCollectionHandle<TDocument> Collection<TDocument>(
CultCache cache,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (cache == null) throw new ArgumentNullException(nameof(cache));
var descriptor = cache.Registry.GetRequired<TDocument>();
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<TDocument>(
CollectionId(collectionId, descriptor),
() => Task.FromResult<IReadOnlyList<TDocument>>(cache.GetAll<TDocument>().ToArray()),
() => cache.Watch<TDocument>().Select(change => ToCollectionChange(change, descriptor.SchemaId)),
sourceList,
route);
}
/// <summary>
/// Creates a typed collection handle over the local CultCache record with the supplied CultName.
/// </summary>
public static CultMeshCollectionHandle<TDocument> CollectionByName<TDocument>(
CultCache cache,
string name,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? 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<TDocument>();
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<TDocument>(
resolvedId,
() => Task.FromResult<IReadOnlyList<TDocument>>(Optional(cache.GetByName<TDocument>(name))),
() => cache.Watch<TDocument>()
.Where(change => MatchesName(cache, name, change))
.Select(change => ToCollectionChange(change, descriptor.SchemaId)),
sourceList,
route);
}
/// <summary>
/// Creates a typed collection handle over local CultCache records matched by an indexed value.
/// </summary>
public static CultMeshCollectionHandle<TDocument> CollectionByIndex<TDocument>(
CultCache cache,
string alias,
string value,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? 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<TDocument>();
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<TDocument>(
resolvedId,
() => Task.FromResult<IReadOnlyList<TDocument>>(Optional(cache.GetByIndex<TDocument>(alias, value))),
() => cache.Watch<TDocument>()
.Where(change => MatchesIndex(cache, alias, value, change))
.Select(change => ToCollectionChange(change, descriptor.SchemaId)),
sourceList,
route);
}
/// <summary>
/// Creates a typed collection handle over all distributed CultNet database records assignable to the document type.
/// </summary>
public static CultMeshCollectionHandle<TDocument> Collection<TDocument>(
CultNetDatabase database,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (database == null) throw new ArgumentNullException(nameof(database));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
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<TDocument>(
CollectionId(collectionId, descriptor),
() => Task.FromResult<IReadOnlyList<TDocument>>(database.GetAll<TDocument>().ToArray()),
() => database.Watch<TDocument>().Select(ToCollectionChange),
sourceList,
route);
}
/// <summary>
/// Creates a typed collection handle over the distributed CultNet database record with the supplied CultName.
/// </summary>
public static CultMeshCollectionHandle<TDocument> CollectionByName<TDocument>(
CultNetDatabase database,
string name,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? 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<TDocument>();
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<TDocument>(
resolvedId,
() => Task.FromResult<IReadOnlyList<TDocument>>(Optional(database.GetByName<TDocument>(name))),
() => database.WatchByName<TDocument>(name).Select(ToCollectionChange),
sourceList,
route);
}
/// <summary>
/// Creates a typed collection handle over distributed CultNet database records matched by an indexed value.
/// </summary>
public static CultMeshCollectionHandle<TDocument> CollectionByIndex<TDocument>(
CultNetDatabase database,
string alias,
string value,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? 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<TDocument>();
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<TDocument>(
resolvedId,
() => Task.FromResult<IReadOnlyList<TDocument>>(Optional(database.GetByIndex<TDocument>(alias, value))),
() => database.WatchByIndex<TDocument>(alias, value).Select(ToCollectionChange),
sourceList,
route);
}
/// <summary>
/// Creates a typed collection handle over all CultMesh node database records assignable to the document type.
/// </summary>
public static CultMeshCollectionHandle<TDocument> Collection<TDocument>(
CultMeshNode node,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (node == null) throw new ArgumentNullException(nameof(node));
return Collection<TDocument>(node.Database, collectionId, sources, routeHint);
}
/// <summary>
/// Creates a typed collection handle over the CultMesh node database record with the supplied CultName.
/// </summary>
public static CultMeshCollectionHandle<TDocument> CollectionByName<TDocument>(
CultMeshNode node,
string name,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (node == null) throw new ArgumentNullException(nameof(node));
return CollectionByName<TDocument>(node.Database, name, collectionId, sources, routeHint);
}
/// <summary>
/// Creates a typed collection handle over CultMesh node database records matched by an indexed value.
/// </summary>
public static CultMeshCollectionHandle<TDocument> CollectionByIndex<TDocument>(
CultMeshNode node,
string alias,
string value,
string? collectionId = null,
IEnumerable<CultMeshProjectionSource>? sources = null,
CultMeshRouteHint? routeHint = null)
where TDocument : class
{
if (node == null) throw new ArgumentNullException(nameof(node));
return CollectionByIndex<TDocument>(node.Database, alias, value, collectionId, sources, routeHint);
}
/// <summary>
/// Describes a typed live feed surface for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshLiveFeedDiagnostic DescribeLiveFeed<TParameters, TResult>(
CultMeshLiveFeed<TParameters, TResult> feed)
{
if (feed == null) throw new ArgumentNullException(nameof(feed));
return new CultMeshLiveFeedDiagnostic(feed.FeedId, feed.RouteHint, feed.Sources);
}
/// <summary>
/// Describes a typed query surface for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshQuerySurfaceDiagnostic DescribeQuerySurface<TParameters, TResult>(
CultMeshQuerySurface<TParameters, TResult> query)
{
if (query == null) throw new ArgumentNullException(nameof(query));
return new CultMeshQuerySurfaceDiagnostic(query.QueryId, query.RouteHint, query.Sources);
}
/// <summary>
/// Describes a typed operation handle for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshOperationHandleDiagnostic DescribeOperationHandle<TRequest, TResponse>(
CultMeshOperationHandle<TRequest, TResponse> operation)
{
if (operation == null) throw new ArgumentNullException(nameof(operation));
return new CultMeshOperationHandleDiagnostic(operation.OperationId);
}
/// <summary>
/// Describes a typed state pointer for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshStatePointerDiagnostic DescribeStatePointer<TValue>(
CultMeshStatePointer<TValue> pointer)
{
if (pointer == null) throw new ArgumentNullException(nameof(pointer));
return new CultMeshStatePointerDiagnostic(pointer.PointerId, pointer.RouteHint, pointer.Sources);
}
/// <summary>
/// Describes a mutable typed state pointer for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshStatePointerDiagnostic DescribeStatePointer<TValue>(
CultMeshMutableStatePointer<TValue> pointer)
{
if (pointer == null) throw new ArgumentNullException(nameof(pointer));
return new CultMeshStatePointerDiagnostic(pointer.PointerId, pointer.RouteHint, pointer.Sources);
}
/// <summary>
/// Describes a native slice view for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshNativeSliceViewDiagnostic DescribeNativeSliceView(
CultMeshNativeSliceViewDescriptor view)
{
return new CultMeshNativeSliceViewDiagnostic(view);
}
/// <summary>
/// Describes a typed projection recipe for tooling, UI, and diagnostics.
/// </summary>
public static CultMeshProjectionRecipeDiagnostic DescribeProjectionRecipe<TParameters, TResult>(
CultMeshProjectionRecipe<TParameters, TResult> recipe)
{
if (recipe == null) throw new ArgumentNullException(nameof(recipe));
return new CultMeshProjectionRecipeDiagnostic(recipe.ProjectionId, recipe.RouteHint, recipe.Sources);
}
/// <summary>
/// Describes a typed query surface as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TParameters, TResult>(
CultMeshQuerySurface<TParameters, TResult> query)
{
if (query == null) throw new ArgumentNullException(nameof(query));
return new CultMeshSurfaceDiagnostic(CultMeshSurfaceKind.Query, query.QueryId, query.RouteHint, query.Sources);
}
/// <summary>
/// Describes a typed operation handle as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TRequest, TResponse>(
CultMeshOperationHandle<TRequest, TResponse> operation)
{
if (operation == null) throw new ArgumentNullException(nameof(operation));
return new CultMeshSurfaceDiagnostic(CultMeshSurfaceKind.Operation, operation.OperationId);
}
/// <summary>
/// Describes a typed projection recipe as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TParameters, TResult>(
CultMeshProjectionRecipe<TParameters, TResult> recipe)
{
if (recipe == null) throw new ArgumentNullException(nameof(recipe));
return new CultMeshSurfaceDiagnostic(CultMeshSurfaceKind.ProjectionRecipe, recipe.ProjectionId, recipe.RouteHint, recipe.Sources);
}
/// <summary>
/// Describes a typed live feed as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TParameters, TResult>(
CultMeshLiveFeed<TParameters, TResult> feed)
{
if (feed == null) throw new ArgumentNullException(nameof(feed));
return new CultMeshSurfaceDiagnostic(CultMeshSurfaceKind.LiveFeed, feed.FeedId, feed.RouteHint, feed.Sources);
}
/// <summary>
/// Describes a typed document handle as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface(
ICultMeshDocumentHandle document)
{
if (document == null) throw new ArgumentNullException(nameof(document));
return new CultMeshSurfaceDiagnostic(
CultMeshSurfaceKind.Document,
document.DocumentId,
document.RouteHint,
document.Sources);
}
/// <summary>
/// Describes a typed collection handle as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TDocument>(
CultMeshCollectionHandle<TDocument> collection)
where TDocument : class
{
if (collection == null) throw new ArgumentNullException(nameof(collection));
return new CultMeshSurfaceDiagnostic(
CultMeshSurfaceKind.Collection,
collection.CollectionId,
collection.RouteHint,
collection.Sources);
}
/// <summary>
/// Describes a typed state pointer as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TValue>(
CultMeshStatePointer<TValue> pointer)
{
if (pointer == null) throw new ArgumentNullException(nameof(pointer));
return new CultMeshSurfaceDiagnostic(
CultMeshSurfaceKind.StatePointer,
pointer.PointerId,
pointer.RouteHint,
pointer.Sources);
}
/// <summary>
/// Describes a mutable typed state pointer as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface<TValue>(
CultMeshMutableStatePointer<TValue> pointer)
{
if (pointer == null) throw new ArgumentNullException(nameof(pointer));
return new CultMeshSurfaceDiagnostic(
CultMeshSurfaceKind.StatePointer,
pointer.PointerId,
pointer.RouteHint,
pointer.Sources);
}
/// <summary>
/// Describes a native slice view as one entry in a surface catalog.
/// </summary>
public static CultMeshSurfaceDiagnostic DescribeSurface(
CultMeshNativeSliceViewDescriptor view)
{
if (view == null) throw new ArgumentNullException(nameof(view));
return new CultMeshSurfaceDiagnostic(CultMeshSurfaceKind.NativeSliceView, view.ViewId, view.Route);
}
/// <summary>
/// Creates a typed surface catalog diagnostic for tooling, UI, and generated bindings.
/// </summary>
public static CultMeshSurfaceCatalogDiagnostic DescribeSurfaceCatalog(
string catalogId,
IEnumerable<CultMeshSurfaceDiagnostic> surfaces)
{
return new CultMeshSurfaceCatalogDiagnostic(catalogId, surfaces);
}
/// <summary>
/// Creates a kind-indexed surface catalog diagnostic for generated bindings, UI, and tools.
/// </summary>
public static CultMeshSurfaceCatalogIndexDiagnostic DescribeSurfaceCatalogIndex(
CultMeshSurfaceCatalogDiagnostic catalog)
{
if (catalog == null) throw new ArgumentNullException(nameof(catalog));
return catalog.IndexByKind();
}
/// <summary>
/// Creates a timer-backed query watcher for runtimes that do not yet have native reactive transport.
/// </summary>
public static Func<TParameters, CultMeshQueryContext, R3.Observable<TResult>> PollingQueryWatcher<TParameters, TResult>(
Func<TParameters, CultMeshQueryContext, Task<TResult>> sample,
CultMeshPollingWatchOptions<TResult>? options = null)
{
if (sample == null) throw new ArgumentNullException(nameof(sample));
var resolvedOptions = options ?? new CultMeshPollingWatchOptions<TResult>();
return (parameters, context) => R3.Observable.Create<TResult>(observer =>
{
var watcher = new CultMeshPollingWatcher<TParameters, TResult>(
sample,
parameters,
context,
resolvedOptions);
var subscription = watcher.Observable.Subscribe(observer);
return new CultMeshCompositeDisposable(subscription, watcher);
});
}
private static async Task<CultNetSnapshotResponseRawMessage> RequestPeerSnapshotAsync<TDocument>(
Func<ICultNetSchemaClient> createClient,
string endpoint,
string recordKey,
CultMeshQueryContext context,
CultMeshPeerSnapshotDocumentOptions options)
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
var selection = new CultNetSelection
{
Schemas = options.Selection?.Schemas ?? new[] { descriptor.SchemaId },
Keys = new[] { recordKey }
};
return await FetchSnapshotAsync(
createClient,
endpoint,
new CultMeshSnapshotRequestOptions
{
Selection = selection,
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
})
.ConfigureAwait(false);
}
// One exact read (docs/cultnet-selection-cut.md, section 4/7): the record this recordKey
// names, decoded as TDocument. Candidates at that key are searched in one order - exact
// schema, then a declared alias of it, then (a foreign/runtime-generated schema id this
// runtime does not recognize by any alias) its payload's own embedded schema stamp,
// alias-matched against TDocument's descriptor. The parse step
// (CultNetDocumentRegistry.TryReadSchemaVersion) is the one decode rule shared with
// DecodeSnapshotDocuments; the match itself stays a direct alias-match against the caller's
// own descriptor rather than a global "which registered type owns this schema string" lookup,
// because that lookup is ambiguous when two document types alias the same schema id (as
// MeshNoteDocument/MeshNoteAliasDocument do in tests/GameCult.Mesh.Tests). No re-filter of the
// server's answer by schema alone, and no fallback that returns a record under the wrong key.
private static TDocument ReadDocumentFromSnapshotResponse<TDocument>(
CultNetSnapshotResponseRawMessage response,
string schemaId,
string recordKey)
where TDocument : class
{
if (response == null) throw new ArgumentNullException(nameof(response));
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
CultNetRawDocumentRecord? exact = null;
CultNetRawDocumentRecord? aliased = null;
CultNetRawDocumentRecord? byPayload = null;
foreach (var candidate in response.Documents)
{
if (!string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal))
continue;
if (exact == null && string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal))
exact = candidate;
else if (aliased == null && CultNetSchemaAliasMatching.Matches(candidate.SchemaId, descriptor))
aliased = candidate;
else if (byPayload == null && CultNetDocumentRegistry.PayloadMatchesSchema(candidate.Payload, descriptor))
byPayload = candidate;
}
var record = exact ?? aliased ?? byPayload;
if (record == null)
{
throw new InvalidOperationException(
$"CultNet snapshot response '{response.MessageId}' did not contain schema '{schemaId}' record '{recordKey}'.");
}
if (!string.Equals(record.PayloadEncoding, "messagepack", StringComparison.Ordinal))
{
throw new InvalidOperationException(
$"CultNet raw document payloadEncoding must be \"messagepack\", not \"{record.PayloadEncoding}\".");
}
return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TDocument), record.Payload);
}
/// <summary>
/// Parses a rudp://host:port endpoint into its host/port parts.
/// </summary>
public static CultMeshRudpEndpoint ParseRudpEndpoint(string endpoint)
{
if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
if (!Uri.TryCreate(endpoint, UriKind.Absolute, out var uri) ||
!string.Equals(uri.Scheme, "rudp", StringComparison.OrdinalIgnoreCase))
{
throw new ArgumentException("RUDP endpoint must use the rudp:// scheme.", nameof(endpoint));
}
if (string.IsNullOrWhiteSpace(uri.Host) || uri.Port <= 0 || uri.Port > 65535)
{
throw new ArgumentException("RUDP endpoint must include a host and port.", nameof(endpoint));
}
var host = uri.Host;
var uriHost = host.Contains(':', StringComparison.Ordinal) && !host.StartsWith("[", StringComparison.Ordinal)
? $"[{host}]"
: host;
return new CultMeshRudpEndpoint(host, uri.Port, $"rudp://{uriHost}:{uri.Port}");
}
/// <summary>
/// Opens one exact hot-body subscription and advertises every body plane this runtime can
/// actually open. The provider derives locality and publishes only the selected plane.
/// </summary>
public static Task<IReadOnlyList<object>> SubscribeHotBodyAsync(
CultNetDatabaseSubscriptionClient subscriptions,
IEnumerable<CultMeshBodyTransportKind> supportedBodyTransports,
CultMeshHotBodySubscription subscription,
CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.Live)
{
if (subscriptions == null) throw new ArgumentNullException(nameof(subscriptions));
if (supportedBodyTransports == null) throw new ArgumentNullException(nameof(supportedBodyTransports));
if (subscription == null) throw new ArgumentNullException(nameof(subscription));
var transports = supportedBodyTransports.Distinct().ToArray();
if (transports.Length == 0)
throw new ArgumentException("At least one readable body transport is required.", nameof(supportedBodyTransports));
return subscriptions.SubscribeAsync(
subscription.SubscriptionId,
recordKeys: subscription.Selection.Keys,
schemaIds: subscription.Selection.Schemas,
consumerRuntimeId: subscription.ConsumerRuntimeId,
bodyIds: new[] { subscription.BodyId },
supportedBodyTransports: transports.Select(value => value.ToString()),
deliveryMode: deliveryMode);
}
/// <summary>Opens an exact hot-body subscription using a resolver's readable planes.</summary>
public static Task<IReadOnlyList<object>> SubscribeHotBodyAsync(
CultNetDatabaseSubscriptionClient subscriptions,
CultMeshBodyPublicationResolver bodyResolver,
CultMeshHotBodySubscription subscription,
CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.Live)
{
if (bodyResolver == null) throw new ArgumentNullException(nameof(bodyResolver));
return SubscribeHotBodyAsync(subscriptions, bodyResolver.SupportedTransports, subscription, deliveryMode);
}
/// <summary>
/// Subscribes to one logical body's latest descriptor as ephemeral control state and declares
/// demand for the body itself. Body bytes are opened separately through the fastest valid plane.
/// </summary>
public static async Task<CultMeshLiveBody> SubscribeLiveBodyAsync(
CultNetDatabaseSubscriptionClient subscriptions,
CultMeshBodyPublicationResolver bodyResolver,
CultMeshLiveBodySubscription subscription,
CancellationToken cancellationToken = default)
{
if (subscriptions == null) throw new ArgumentNullException(nameof(subscriptions));
if (bodyResolver == null) throw new ArgumentNullException(nameof(bodyResolver));
if (subscription == null) throw new ArgumentNullException(nameof(subscription));
var transports = bodyResolver.SupportedTransports.Distinct().ToArray();
if (transports.Length == 0)
throw new InvalidOperationException("The body resolver has no readable transport planes.");
var liveBody = new CultMeshLiveBody(subscriptions, bodyResolver, subscription);
try
{
var initial = await subscriptions.SubscribeAsync(
subscription.SubscriptionId,
recordKeys: new[] { subscription.PublicationRecordKey },
schemaIds: new[] { CultMeshBodyPublicationSchemaVersions.Publication },
consumerRuntimeId: subscription.ConsumerRuntimeId,
bodyIds: new[] { subscription.BodyId },
supportedBodyTransports: transports.Select(value => value.ToString()),
deliveryMode: CultNetDatabaseSubscriptionDeliveryMode.Live,
cancellationToken: cancellationToken)
.ConfigureAwait(false);
var publications = initial.OfType<CultMeshBodyPublicationDocument>().ToArray();
if (publications.Length > 1)
throw new InvalidOperationException(
$"Live body subscription '{subscription.SubscriptionId}' expected at most one publication for " +
$"'{subscription.BodyId}' but received {publications.Length}.");
if (publications.Length == 1)
liveBody.Initialize(publications[0]);
return liveBody;
}
catch
{
liveBody.Dispose();
throw;
}
}
/// <summary>Resolves a CultMesh URI or RUDP endpoint into a concrete RUDP endpoint.</summary>
public static CultMeshRudpEndpoint ResolveRudpEndpoint(string endpointOrUri)
{
if (string.IsNullOrWhiteSpace(endpointOrUri))
throw new ArgumentException("Value must be non-empty.", nameof(endpointOrUri));
if (!Uri.TryCreate(endpointOrUri, UriKind.Absolute, out var uri))
return ParseRudpEndpoint(NormalizeBareRudpEndpoint(endpointOrUri));
if (string.Equals(uri.Scheme, "rudp", StringComparison.OrdinalIgnoreCase))
return ParseRudpEndpoint(endpointOrUri);
if (!string.Equals(uri.Scheme, "cultmesh", StringComparison.OrdinalIgnoreCase))
throw new ArgumentException("RUDP endpoint must use the cultmesh:// or rudp:// scheme.", nameof(endpointOrUri));
if (string.IsNullOrWhiteSpace(uri.Host))
throw new ArgumentException("CultMesh URI must include an authority.", nameof(endpointOrUri));
var authorityKey = MakeCultMeshAuthorityEnvironmentKey(uri.Host);
var endpoint = Environment.GetEnvironmentVariable($"CULTMESH_URI_{authorityKey}_RUDP") ??
Environment.GetEnvironmentVariable($"{authorityKey}_CULTMESH_RUDP_ENDPOINT");
if (string.IsNullOrWhiteSpace(endpoint))
throw new InvalidOperationException(
$"CultMesh URI '{endpointOrUri}' is unresolved. Set CULTMESH_URI_{authorityKey}_RUDP to a rudp:// endpoint.");
return ParseRudpEndpoint(endpoint);
}
private static string NormalizeBareRudpEndpoint(string endpoint)
{
var trimmed = endpoint.Trim();
return trimmed.Contains("://", StringComparison.Ordinal) ? trimmed : $"rudp://{trimmed}";
}
private static string MakeCultMeshAuthorityEnvironmentKey(string authority)
{
var chars = authority.Select(static value =>
((value >= 'a' && value <= 'z') || (value >= 'A' && value <= 'Z') || (value >= '0' && value <= '9'))
? char.ToUpperInvariant(value)
: '_');
var key = string.Concat(chars).Trim('_');
return string.IsNullOrWhiteSpace(key) ? "ODIN" : key;
}
/// <summary>
/// Creates a CultNet RUDP server transport with CultMesh-branded defaults.
/// </summary>
public static CultNetRudpSocketTransportConnection CreateRudpServer(
string runtimeId,
uint connectionId,
CultMeshRudpSocketOptions? options = null)
{
options ??= new CultMeshRudpSocketOptions();
return new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = runtimeId,
Socket = options.Socket ?? BindRudpSocket(options.BindHost, options.BindPort),
Mode = CultNetRudpSocketMode.Server,
ConnectionId = connectionId,
InitialSequence = options.InitialSequence,
ResendDelayMs = options.ResendDelayMs,
TransportId = options.TransportId,
MaxPayloadBytes = options.MaxPayloadBytes,
MaxFragmentBytes = options.MaxFragmentBytes,
MaxPendingReliablePackets = options.MaxPendingReliablePackets,
ReconnectPolicy = options.ReconnectPolicy
});
}
/// <summary>
/// Creates a CultNet RUDP client transport for a parsed endpoint.
/// </summary>
public static CultNetRudpSocketTransportConnection CreateRudpClient(
string runtimeId,
uint connectionId,
CultMeshRudpEndpoint endpoint,
CultMeshRudpSocketOptions? options = null)
{
if (endpoint == null) throw new ArgumentNullException(nameof(endpoint));
options ??= new CultMeshRudpSocketOptions();
return new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = runtimeId,
Socket = options.Socket ?? BindRudpSocket(options.BindHost, options.BindPort),
Mode = CultNetRudpSocketMode.Client,
RemoteEndPoint = new IPEndPoint(ResolveRudpAddress(endpoint.Host), endpoint.Port),
ConnectionId = connectionId,
InitialSequence = options.InitialSequence,
ResendDelayMs = options.ResendDelayMs,
TransportId = options.TransportId,
MaxPayloadBytes = options.MaxPayloadBytes,
MaxFragmentBytes = options.MaxFragmentBytes,
MaxPendingReliablePackets = options.MaxPendingReliablePackets,
ReconnectPolicy = options.ReconnectPolicy
});
}
/// <summary>
/// Creates a CultNet RUDP client transport for a rudp://host:port endpoint.
/// </summary>
public static CultNetRudpSocketTransportConnection CreateRudpClient(
string runtimeId,
uint connectionId,
string endpoint,
CultMeshRudpSocketOptions? options = null)
{
return CreateRudpClient(runtimeId, connectionId, ResolveRudpEndpoint(endpoint), options);
}
/// <summary>
/// Creates a CultNet RUDP client transport from a peer-card contact hint.
/// </summary>
public static CultNetRudpSocketTransportConnection CreateRudpClientForPeer(
string runtimeId,
uint connectionId,
CultMeshPeerCard peer,
CultMeshRudpSocketOptions? options = null)
{
if (peer == null) throw new ArgumentNullException(nameof(peer));
var endpoint = peer.Endpoints.FirstOrDefault(value => value.StartsWith("rudp://", StringComparison.OrdinalIgnoreCase));
if (endpoint == null)
{
throw new InvalidOperationException($"Peer {peer.PeerId} does not advertise a RUDP endpoint.");
}
return CreateRudpClient(runtimeId, connectionId, endpoint, options);
}
/// <summary>
/// Creates a CultNet RUDP client transport from the first peer authorized for a Verse role.
/// </summary>
[Obsolete("Supply CultMeshAuthorityResolver and an explicit authority epoch. This compatibility path denies unverifiable leases.")]
public static CultNetRudpSocketTransportConnection CreateRudpClientForAuthorizedPeer(
string runtimeId,
uint connectionId,
CultMeshPeerCatalog peers,
CultMeshAuthorityLeaseCatalog leases,
string verseId,
string role,
string? shardId = null,
DateTimeOffset? at = null,
CultMeshRudpSocketOptions? options = null)
{
if (peers == null) throw new ArgumentNullException(nameof(peers));
if (leases == null) throw new ArgumentNullException(nameof(leases));
var resolver = CultMeshAuthorityResolver.CreateDenyByDefault(leases,
at.HasValue ? new CultMeshFixedAuthorityClock(at.Value) : null);
return CreateRudpClientForAuthorizedPeer(runtimeId, connectionId, peers, resolver, 0, verseId, role, shardId, null, options);
}
/// <summary>Creates a RUDP client from the first contact accepted by the authority resolver.</summary>
public static CultNetRudpSocketTransportConnection CreateRudpClientForAuthorizedPeer(
string runtimeId,
uint connectionId,
CultMeshPeerCatalog peers,
CultMeshAuthorityResolver authority,
long authorityEpoch,
string verseId,
string role,
string? shardId = null,
string? resourceScope = null,
CultMeshRudpSocketOptions? options = null)
{
if (peers == null) throw new ArgumentNullException(nameof(peers));
if (authority == null) throw new ArgumentNullException(nameof(authority));
var peer = peers.FirstAuthorized(verseId, role, authority, authorityEpoch, shardId, resourceScope);
if (peer == null)
{
throw new InvalidOperationException($"No authorized RUDP peer for role {role} in Verse {verseId}.");
}
return CreateRudpClientForPeer(runtimeId, connectionId, peer, options);
}
/// <summary>
/// Creates and handshakes a CultNet RUDP client transport for a parsed endpoint.
/// </summary>
public static CultNetRudpSocketTransportConnection ConnectRudpClient(
string runtimeId,
uint connectionId,
CultMeshRudpEndpoint endpoint,
CultMeshRudpClientOptions? options = null)
{
options ??= new CultMeshRudpClientOptions();
var client = CreateRudpClient(runtimeId, connectionId, endpoint, options.SocketOptions);
if (!client.ConnectAndWait(options.ConnectPayload, options.ConnectTimeout, options.PollInterval))
{
client.Dispose();
throw new TimeoutException($"Timed out waiting for RUDP client {runtimeId} to connect.");
}
return client;
}
/// <summary>
/// Creates and handshakes a CultNet RUDP client transport for a rudp://host:port endpoint.
/// </summary>
public static CultNetRudpSocketTransportConnection ConnectRudpClient(
string runtimeId,
uint connectionId,
string endpoint,
CultMeshRudpClientOptions? options = null)
{
return ConnectRudpClient(runtimeId, connectionId, ParseRudpEndpoint(endpoint), options);
}
/// <summary>
/// Creates and handshakes a CultNet RUDP client transport from a peer-card contact hint.
/// </summary>
public static CultNetRudpSocketTransportConnection ConnectRudpClientForPeer(
string runtimeId,
uint connectionId,
CultMeshPeerCard peer,
CultMeshRudpClientOptions? options = null)
{
if (peer == null) throw new ArgumentNullException(nameof(peer));
var endpoint = peer.Endpoints.FirstOrDefault(value => value.StartsWith("rudp://", StringComparison.OrdinalIgnoreCase));
if (endpoint == null)
{
throw new InvalidOperationException($"Peer {peer.PeerId} does not advertise a RUDP endpoint.");
}
return ConnectRudpClient(runtimeId, connectionId, endpoint, options);
}
/// <summary>
/// Creates and handshakes a CultNet RUDP client transport from the first peer authorized for a Verse role.
/// </summary>
[Obsolete("Supply CultMeshAuthorityResolver and an explicit authority epoch. This compatibility path denies unverifiable leases.")]
public static CultNetRudpSocketTransportConnection ConnectRudpClientForAuthorizedPeer(
string runtimeId,
uint connectionId,
CultMeshPeerCatalog peers,
CultMeshAuthorityLeaseCatalog leases,
string verseId,
string role,
string? shardId = null,
DateTimeOffset? at = null,
CultMeshRudpClientOptions? options = null)
{
if (peers == null) throw new ArgumentNullException(nameof(peers));
if (leases == null) throw new ArgumentNullException(nameof(leases));
var resolver = CultMeshAuthorityResolver.CreateDenyByDefault(leases,
at.HasValue ? new CultMeshFixedAuthorityClock(at.Value) : null);
return ConnectRudpClientForAuthorizedPeer(runtimeId, connectionId, peers, resolver, 0, verseId, role, shardId, null, options);
}
/// <summary>Creates and handshakes a RUDP client from the first contact accepted by the authority resolver.</summary>
public static CultNetRudpSocketTransportConnection ConnectRudpClientForAuthorizedPeer(
string runtimeId,
uint connectionId,
CultMeshPeerCatalog peers,
CultMeshAuthorityResolver authority,
long authorityEpoch,
string verseId,
string role,
string? shardId = null,
string? resourceScope = null,
CultMeshRudpClientOptions? options = null)
{
if (peers == null) throw new ArgumentNullException(nameof(peers));
if (authority == null) throw new ArgumentNullException(nameof(authority));
var peer = peers.FirstAuthorized(verseId, role, authority, authorityEpoch, shardId, resourceScope);
if (peer == null)
{
throw new InvalidOperationException($"No authorized RUDP peer for role {role} in Verse {verseId}.");
}
return ConnectRudpClientForPeer(runtimeId, connectionId, peer, options);
}
private sealed class CultMeshFixedAuthorityClock : ICultMeshClock
{
public CultMeshFixedAuthorityClock(DateTimeOffset utcNow) => UtcNow = utcNow;
public DateTimeOffset UtcNow { get; }
public System.Threading.Tasks.Task DelayAsync(TimeSpan delay, System.Threading.CancellationToken cancellationToken = default) =>
System.Threading.Tasks.Task.CompletedTask;
}
/// <summary>
/// Creates a committer for writing quorum simulation facts to a mesh database.
/// </summary>
public static CultMeshSimulationFactCommitter CreateSimulationFactCommitter(CultNetDatabase database)
{
return new CultMeshSimulationFactCommitter(database);
}
/// <summary>
/// Creates a gameplay-facing session facade over a CultMesh node.
/// </summary>
public static CultMeshGameSession CreateGameSession(
CultMeshNode node,
CultMeshGameSessionOptions? options = null)
{
return new CultMeshGameSession(node, options);
}
/// <summary>
/// Attaches Verse discovery responses to a CultMesh node.
/// </summary>
public static CultMeshVerseDiscoveryServer ServeVerseCatalog(
CultMeshNode node,
CultMeshVerseCatalog catalog)
{
if (node == null) throw new ArgumentNullException(nameof(node));
return new CultMeshVerseDiscoveryServer(node.Server, catalog);
}
/// <summary>
/// Creates a client for fetching Verse catalogs from discovery endpoints.
/// </summary>
public static CultMeshVerseDiscoveryClient CreateVerseDiscoveryClient(
CultMeshVerseDiscoveryClientOptions? options = null)
{
return new CultMeshVerseDiscoveryClient(options);
}
/// <summary>
/// Attaches peer exchange responses to a CultMesh node.
/// </summary>
public static CultMeshPeerExchangeServer ServePeerExchange(
CultMeshNode node,
CultMeshPeerCatalog catalog)
{
if (node == null) throw new ArgumentNullException(nameof(node));
return new CultMeshPeerExchangeServer(node.Server, catalog);
}
/// <summary>
/// Creates a client for fetching peer cards from peer exchange endpoints.
/// </summary>
public static CultMeshPeerExchangeClient CreatePeerExchangeClient(
CultMeshPeerExchangeClientOptions? options = null)
{
return new CultMeshPeerExchangeClient(options);
}
/// <summary>
/// Creates a CultNet client configured for CultMesh use.
/// </summary>
public static Client CreateClient(ClientSecurityOptions? security = null, Action<Client>? configureClient = null)
{
return CultNetLocal.CreateClient(security, configureClient);
}
/// <summary>
/// Creates and connects a CultNet client to a CultMesh node.
/// </summary>
public static Client ConnectClient(
string host = "localhost",
int port = 3075,
ClientSecurityOptions? security = null,
Action<Client>? configureClient = null)
{
return CultNetLocal.ConnectClient(host, port, security, configureClient);
}
private static Socket BindRudpSocket(string host, int port)
{
var address = ResolveRudpAddress(host);
var socket = new Socket(address.AddressFamily, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(new IPEndPoint(address, port));
socket.ReceiveTimeout = 20;
return socket;
}
private static IPAddress ResolveRudpAddress(string host)
{
if (string.IsNullOrWhiteSpace(host)) throw new ArgumentException("Value must be non-empty.", nameof(host));
if (IPAddress.TryParse(host, out var parsed))
{
return parsed;
}
return Dns.GetHostAddresses(host).FirstOrDefault()
?? throw new InvalidOperationException($"Could not resolve RUDP host {host}.");
}
private static string ResolveDocumentId(string? documentId, CultRecordKey key)
{
return string.IsNullOrWhiteSpace(documentId)
? key.Value
: documentId!;
}
private static string CollectionId(string? collectionId, CultDocumentDescriptor descriptor)
{
return string.IsNullOrWhiteSpace(collectionId)
? descriptor.SchemaId
: collectionId!;
}
private static IReadOnlyList<TDocument> Optional<TDocument>(TDocument? document)
where TDocument : class
{
return document == null
? Array.Empty<TDocument>()
: new[] { document };
}
private static bool MatchesName<TDocument>(
CultCache cache,
string name,
CultCacheDocumentChange<TDocument> change)
where TDocument : class
{
var current = cache.GetByName<TDocument>(name);
return (change.Document != null && ReferenceEquals(current, change.Document)) ||
(change.PreviousDocument != null && ReferenceEquals(current, change.PreviousDocument));
}
private static bool MatchesIndex<TDocument>(
CultCache cache,
string alias,
string value,
CultCacheDocumentChange<TDocument> change)
where TDocument : class
{
var current = cache.GetByIndex<TDocument>(alias, value);
return (change.Document != null && ReferenceEquals(current, change.Document)) ||
(change.PreviousDocument != null && ReferenceEquals(current, change.PreviousDocument));
}
private static CultMeshCollectionChange<TDocument> ToCollectionChange<TDocument>(
CultCacheDocumentChange<TDocument> change,
string schemaId)
where TDocument : class
{
return new CultMeshCollectionChange<TDocument>(
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<TDocument> ToCollectionChange<TDocument>(
CultNetDatabaseChange<TDocument> change)
where TDocument : class
{
return new CultMeshCollectionChange<TDocument>(
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<TDocument>(CultCache cache, CultRecordKey key)
where TDocument : class =>
RequireDocument<TDocument>(cache.Get(key), key);
private static TDocument RequireDocument<TDocument>(object? untyped, CultRecordKey key)
where TDocument : class
{
if (untyped is TDocument document)
return document;
if (untyped != null && IsSameCultDocumentSchema<TDocument>(untyped.GetType()))
return ConvertUntypedDocument<TDocument>(untyped);
throw new KeyNotFoundException(
$"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}.");
}
private static async Task<TDocument> ReadRequiredAsync<TDocument>(
CultNetDatabase database,
CultRecordKey key)
where TDocument : class
{
var document = await database.GetAsync<TDocument>(key).ConfigureAwait(false);
return document
?? throw new KeyNotFoundException(
$"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}.");
}
private static async Task<TDocument> ReadDatabaseDocumentRequiredAsync<TDocument>(
CultNetDatabase database,
CultRecordKey key)
where TDocument : class
{
var document = await database.GetAsync<TDocument>(key).ConfigureAwait(false);
if (document != null)
return document;
var untyped = database.Cache.Get(key);
if (untyped != null && IsSameCultDocumentSchema<TDocument>(untyped.GetType()))
return ConvertUntypedDocument<TDocument>(untyped);
throw new KeyNotFoundException(
$"CultMesh document '{key.Value}' was not found as {typeof(TDocument).FullName}.");
}
private static R3.Observable<TDocument> WatchDatabaseRecordAs<TDocument>(
CultNetDatabase database,
CultRecordKey key)
where TDocument : class
{
var descriptor = CultDocumentRegistry.Shared.GetRequired<TDocument>();
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<TDocument>(
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<TDocument>(current);
return true;
}
private static bool IsSameCultDocumentSchema<TDocument>(Type documentType)
where TDocument : class
{
return IsSameCultDocumentSchema(
documentType,
CultDocumentRegistry.Shared.GetRequired<TDocument>());
}
// The one schema-alias matcher (docs/cultnet-selection-cut.md, D4): every runtime-side
// schema identity decision, wire or local, goes through CultNetSchemaAliasMatching.
private static bool IsSameCultDocumentSchema(
Type documentType,
CultDocumentDescriptor descriptor,
string? schemaId = null)
{
if (!string.IsNullOrWhiteSpace(schemaId) &&
CultNetSchemaAliasMatching.Matches(schemaId, descriptor))
{
return true;
}
CultDocumentDescriptor storedDescriptor;
try
{
storedDescriptor = CultDocumentRegistry.Shared.GetRequired(documentType);
}
catch (Exception)
{
return false;
}
return CultNetSchemaAliasMatching.Matches(storedDescriptor.SchemaVersion, descriptor);
}
private static TDocument ConvertUntypedDocument<TDocument>(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<TDocument>(
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<TDocument>()))
{
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);
}
}
}