using System;
using System.Collections;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Linq.Expressions;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
using MessagePack;
using R3;
namespace GameCult.Mesh
{
///
/// Describes the route CultMesh used or should prefer for a typed operation, query, pointer, or native view.
///
public enum CultMeshLocalityKind
{
/// Let CultMesh choose the fastest valid route.
Automatic,
/// The target is in the same process.
InProcess,
/// The target is available through a shared CultCache slab or native shared memory view.
SharedMemory,
/// The target is reachable through local inter-process transport.
Ipc,
/// The target is reachable through a network Verse route.
Network,
/// The target is hosted in or exposed to a WebAssembly runtime.
Wasm
}
///
/// Names a CultMesh locality decision without exposing transport plumbing to application code.
///
public sealed class CultMeshRouteHint
{
/// Creates a route hint.
public CultMeshRouteHint(CultMeshLocalityKind kind = CultMeshLocalityKind.Automatic, string? description = null)
{
Kind = kind;
Description = description;
}
/// Gets the requested or observed locality kind.
public CultMeshLocalityKind Kind { get; }
/// Gets optional human-facing route diagnostics.
public string? Description { get; }
/// Gets the default automatic route hint.
public static CultMeshRouteHint Automatic { get; } = new();
}
///
/// Flat, transport-friendly fields for a CultMesh route hint.
///
[MessagePackObject]
[MessagePackFormatter(typeof(CultMeshRouteRecordFormatter))]
public sealed class CultMeshRouteRecord
{
/// Creates flat route fields.
[SerializationConstructor]
public CultMeshRouteRecord(string? kind = null, string? description = null)
{
Kind = kind ?? "";
Description = description ?? "";
}
/// Gets the serialized route locality kind.
[Key(0)] public string Kind { get; }
/// Gets the serialized route description.
[Key(1)] public string Description { get; }
/// Creates flat route fields from a route hint.
public static CultMeshRouteRecord FromRoute(CultMeshRouteHint? routeHint)
{
var route = routeHint ?? CultMeshRouteHint.Automatic;
return new CultMeshRouteRecord(route.Kind.ToString(), route.Description);
}
/// Rehydrates flat route fields into a route hint.
public CultMeshRouteHint ToRoute(CultMeshRouteHint? fallback = null)
{
var fallbackRoute = fallback ?? CultMeshRouteHint.Automatic;
return new CultMeshRouteHint(
ParseKind(Kind, fallbackRoute.Kind),
string.IsNullOrWhiteSpace(Description) ? fallbackRoute.Description : Description);
}
private static CultMeshLocalityKind ParseKind(string? value, CultMeshLocalityKind fallback)
{
if (string.IsNullOrWhiteSpace(value))
return fallback;
if (Enum.TryParse(value, ignoreCase: true, out var parsed))
return parsed;
var normalized = value.Replace("-", "", StringComparison.Ordinal);
foreach (CultMeshLocalityKind candidate in Enum.GetValues(typeof(CultMeshLocalityKind)))
{
if (string.Equals(candidate.ToString(), normalized, StringComparison.OrdinalIgnoreCase))
return candidate;
}
return fallback;
}
}
///
/// Shared Verse invocation context used by generated domain sugar.
///
public sealed class CultMeshVerseContext
{
/// Creates a Verse context.
public CultMeshVerseContext(
string verseId,
string runtimeId,
CultMeshRouteHint? routeHint = null,
IEnumerable? claims = null)
{
VerseId = RequireNonEmpty(verseId, nameof(verseId));
RuntimeId = RequireNonEmpty(runtimeId, nameof(runtimeId));
RouteHint = routeHint ?? CultMeshRouteHint.Automatic;
Claims = claims?.ToArray() ?? Array.Empty();
}
/// Gets the semantic Verse id.
public string VerseId { get; }
/// Gets the runtime using the Verse.
public string RuntimeId { get; }
/// Gets the preferred route for generated operations and queries.
public CultMeshRouteHint RouteHint { get; }
/// Gets authority claims carried by generated operations.
public IReadOnlyList Claims { get; }
/// Returns a copy with one additional authority claim.
public CultMeshVerseContext WithClaim(CultMeshAuthorityClaim claim)
{
if (claim == null) throw new ArgumentNullException(nameof(claim));
return new CultMeshVerseContext(VerseId, RuntimeId, RouteHint, Claims.Concat(new[] { claim }));
}
/// Returns a copy with the supplied route hint.
public CultMeshVerseContext WithRoute(CultMeshRouteHint routeHint)
{
if (routeHint == null) throw new ArgumentNullException(nameof(routeHint));
return new CultMeshVerseContext(VerseId, RuntimeId, routeHint, Claims);
}
/// Creates a typed operation context from the Verse context.
public CultMeshOperationContext OperationContext(string? idempotencyKey = null)
{
return new CultMeshOperationContext(RuntimeId, Claims, RouteHint, idempotencyKey);
}
/// Creates a typed query context from the Verse context.
public CultMeshQueryContext QueryContext()
{
return new CultMeshQueryContext(RuntimeId, RouteHint);
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Managed Verse handle that generated domain packages can bind to.
///
public sealed class CultMeshVerse
{
/// Creates a managed Verse handle.
public CultMeshVerse(CultMeshVerseContext context)
{
Context = context ?? throw new ArgumentNullException(nameof(context));
}
/// Gets the Verse context shared by generated domain sugar.
public CultMeshVerseContext Context { get; }
/// Gets the semantic Verse id.
public string VerseId => Context.VerseId;
/// Gets the runtime using the Verse.
public string RuntimeId => Context.RuntimeId;
/// Binds this Verse to a generated domain facade.
public TSchema Use(Func schemaFactory)
{
if (schemaFactory == null) throw new ArgumentNullException(nameof(schemaFactory));
return schemaFactory(Context);
}
/// Returns a Verse handle with one additional authority claim.
public CultMeshVerse WithClaim(CultMeshAuthorityClaim claim)
{
return new CultMeshVerse(Context.WithClaim(claim));
}
/// Returns a Verse handle with the supplied route hint.
public CultMeshVerse WithRoute(CultMeshRouteHint routeHint)
{
return new CultMeshVerse(Context.WithRoute(routeHint));
}
/// Creates a typed operation context from this Verse.
public CultMeshOperationContext OperationContext(string? idempotencyKey = null)
{
return Context.OperationContext(idempotencyKey);
}
/// Creates a typed query context from this Verse.
public CultMeshQueryContext QueryContext()
{
return Context.QueryContext();
}
/// Binds an operation handle to this Verse.
public CultMeshBoundOperationHandle BindOperation(
CultMeshOperationHandle operation)
{
return new CultMeshBoundOperationHandle(Context, operation);
}
/// Binds a query surface to this Verse.
public CultMeshBoundQuerySurface BindQuery(
CultMeshQuerySurface query)
{
return new CultMeshBoundQuerySurface(Context, query);
}
/// Binds a live feed to this Verse.
public CultMeshBoundLiveFeed BindLiveFeed(
CultMeshLiveFeed feed)
{
return new CultMeshBoundLiveFeed(Context, feed);
}
/// Binds a state pointer to this Verse.
public CultMeshBoundStatePointer BindStatePointer(
CultMeshStatePointer pointer)
{
return new CultMeshBoundStatePointer(Context, pointer);
}
/// Binds a mutable state pointer to this Verse.
public CultMeshBoundMutableStatePointer BindMutableStatePointer(
CultMeshMutableStatePointer pointer)
{
return new CultMeshBoundMutableStatePointer(Context, pointer);
}
}
///
/// Describes a runtime authority claim attached to a typed operation.
///
public sealed class CultMeshAuthorityClaim
{
/// Creates an authority claim.
public CultMeshAuthorityClaim(string role, string? shardId = null, string? leaseId = null)
{
Role = RequireNonEmpty(role, nameof(role));
ShardId = shardId;
LeaseId = leaseId;
}
/// Gets the role or claim kind being asserted.
public string Role { get; }
/// Gets the optional shard covered by the claim.
public string? ShardId { get; }
/// Gets the optional lease id backing the claim.
public string? LeaseId { get; }
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Context supplied when invoking a typed CultMesh operation.
///
public sealed class CultMeshOperationContext
{
/// Creates an operation context.
public CultMeshOperationContext(
string runtimeId,
IEnumerable? claims = null,
CultMeshRouteHint? routeHint = null,
string? idempotencyKey = null)
{
RuntimeId = RequireNonEmpty(runtimeId, nameof(runtimeId));
Claims = claims?.ToArray() ?? Array.Empty();
RouteHint = routeHint ?? CultMeshRouteHint.Automatic;
IdempotencyKey = idempotencyKey;
}
/// Gets the runtime invoking the operation.
public string RuntimeId { get; }
/// Gets authority claims attached to the operation.
public IReadOnlyList Claims { get; }
/// Gets the preferred route for the operation.
public CultMeshRouteHint RouteHint { get; }
/// Gets an optional operation idempotency key.
public string? IdempotencyKey { get; }
/// Creates an operation context for a runtime.
public static CultMeshOperationContext ForRuntime(string runtimeId)
{
return new CultMeshOperationContext(runtimeId);
}
/// Returns a copy with one additional authority claim.
public CultMeshOperationContext WithClaim(CultMeshAuthorityClaim claim)
{
if (claim == null) throw new ArgumentNullException(nameof(claim));
return new CultMeshOperationContext(RuntimeId, Claims.Concat(new[] { claim }), RouteHint, IdempotencyKey);
}
/// Returns a copy with the supplied route hint.
public CultMeshOperationContext WithRoute(CultMeshRouteHint routeHint)
{
if (routeHint == null) throw new ArgumentNullException(nameof(routeHint));
return new CultMeshOperationContext(RuntimeId, Claims, routeHint, IdempotencyKey);
}
/// Returns a copy with the supplied idempotency key.
public CultMeshOperationContext WithIdempotencyKey(string idempotencyKey)
{
return new CultMeshOperationContext(
RuntimeId,
Claims,
RouteHint,
RequireNonEmpty(idempotencyKey, nameof(idempotencyKey)));
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Receipt returned by a typed CultMesh operation.
///
public sealed class CultMeshOperationReceipt
{
/// Creates an operation receipt.
public CultMeshOperationReceipt(
string operationId,
bool accepted,
CultMeshRouteHint? route = null,
string? diagnostic = null)
{
OperationId = RequireNonEmpty(operationId, nameof(operationId));
Accepted = accepted;
Route = route ?? CultMeshRouteHint.Automatic;
Diagnostic = diagnostic;
}
/// Gets the semantic operation id.
public string OperationId { get; }
/// Gets whether the operation was accepted by the target authority boundary.
public bool Accepted { get; }
/// Gets the route used for the operation.
public CultMeshRouteHint Route { get; }
/// Gets optional operation diagnostics.
public string? Diagnostic { get; }
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Method-shaped handle for a typed CultMesh operation.
///
public sealed class CultMeshOperationHandle
{
private readonly Func> _invoke;
/// Creates an operation handle.
public CultMeshOperationHandle(
string operationId,
Func> invoke)
{
OperationId = RequireNonEmpty(operationId, nameof(operationId));
_invoke = invoke ?? throw new ArgumentNullException(nameof(invoke));
}
/// Gets the semantic operation id.
public string OperationId { get; }
/// Invokes the typed operation.
public Task InvokeAsync(TRequest request, CultMeshOperationContext context)
{
if (context == null) throw new ArgumentNullException(nameof(context));
return _invoke(request, context);
}
/// Invokes the typed operation for one runtime using default context.
public Task InvokeAsync(TRequest request, string runtimeId)
{
return InvokeAsync(request, CultMeshOperationContext.ForRuntime(runtimeId));
}
/// Binds this operation to a Verse context.
public CultMeshBoundOperationHandle Bind(CultMeshVerseContext context)
{
return new CultMeshBoundOperationHandle(context, this);
}
/// Binds this operation to a Verse.
public CultMeshBoundOperationHandle Bind(CultMeshVerse verse)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Bind(verse.Context);
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Operation handle pre-bound to a Verse so generated domain sugar can invoke without context plumbing.
///
public sealed class CultMeshBoundOperationHandle
{
/// Creates a Verse-bound operation handle.
public CultMeshBoundOperationHandle(
CultMeshVerseContext context,
CultMeshOperationHandle operation)
{
Context = context ?? throw new ArgumentNullException(nameof(context));
Operation = operation ?? throw new ArgumentNullException(nameof(operation));
}
/// Gets the bound Verse context.
public CultMeshVerseContext Context { get; }
/// Gets the underlying operation handle.
public CultMeshOperationHandle Operation { get; }
/// Gets the semantic operation id.
public string OperationId => Operation.OperationId;
/// Invokes the operation through the bound Verse.
public Task InvokeAsync(TRequest request, string? idempotencyKey = null)
{
return Operation.InvokeAsync(request, Context.OperationContext(idempotencyKey));
}
}
///
/// Context supplied when executing a typed CultMesh query surface.
///
public sealed class CultMeshQueryContext
{
/// Creates a query context.
public CultMeshQueryContext(string runtimeId, CultMeshRouteHint? routeHint = null)
{
RuntimeId = RequireNonEmpty(runtimeId, nameof(runtimeId));
RouteHint = routeHint ?? CultMeshRouteHint.Automatic;
}
/// Gets the runtime executing the query.
public string RuntimeId { get; }
/// Gets the preferred route for the query.
public CultMeshRouteHint RouteHint { get; }
/// Creates a query context for a runtime.
public static CultMeshQueryContext ForRuntime(string runtimeId)
{
return new CultMeshQueryContext(runtimeId);
}
/// Returns a copy with the supplied route hint.
public CultMeshQueryContext WithRoute(CultMeshRouteHint routeHint)
{
if (routeHint == null) throw new ArgumentNullException(nameof(routeHint));
return new CultMeshQueryContext(RuntimeId, routeHint);
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Fluent builder for typed operation context. This is sugar over ,
/// intended for call sites that should read like native Verse operations instead of transport setup.
///
public sealed class CultMeshOperationContextBuilder
{
private readonly string _runtimeId;
private readonly List _claims = new();
private CultMeshRouteHint _routeHint = CultMeshRouteHint.Automatic;
private string? _idempotencyKey;
/// Creates a builder for one runtime.
public CultMeshOperationContextBuilder(string runtimeId)
{
_runtimeId = RequireNonEmpty(runtimeId, nameof(runtimeId));
}
/// Adds an authority claim.
public CultMeshOperationContextBuilder Claim(string role, string? shardId = null, string? leaseId = null)
{
_claims.Add(new CultMeshAuthorityClaim(role, shardId, leaseId));
return this;
}
/// Adds an authority claim.
public CultMeshOperationContextBuilder Claim(CultMeshAuthorityClaim claim)
{
_claims.Add(claim ?? throw new ArgumentNullException(nameof(claim)));
return this;
}
/// Adds many authority claims.
public CultMeshOperationContextBuilder Claims(IEnumerable claims)
{
if (claims == null) throw new ArgumentNullException(nameof(claims));
_claims.AddRange(claims);
return this;
}
/// Sets the preferred route hint.
public CultMeshOperationContextBuilder Route(CultMeshLocalityKind kind, string? description = null)
{
_routeHint = new CultMeshRouteHint(kind, description);
return this;
}
/// Sets the operation idempotency key.
public CultMeshOperationContextBuilder Idempotency(string idempotencyKey)
{
_idempotencyKey = RequireNonEmpty(idempotencyKey, nameof(idempotencyKey));
return this;
}
/// Creates the immutable operation context.
public CultMeshOperationContext Build()
{
return new CultMeshOperationContext(_runtimeId, _claims, _routeHint, _idempotencyKey);
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Fluent builder for typed query context.
///
public sealed class CultMeshQueryContextBuilder
{
private readonly string _runtimeId;
private CultMeshRouteHint _routeHint = CultMeshRouteHint.Automatic;
/// Creates a builder for one runtime.
public CultMeshQueryContextBuilder(string runtimeId)
{
_runtimeId = RequireNonEmpty(runtimeId, nameof(runtimeId));
}
/// Sets the preferred route hint.
public CultMeshQueryContextBuilder Route(CultMeshLocalityKind kind, string? description = null)
{
_routeHint = new CultMeshRouteHint(kind, description);
return this;
}
/// Creates the immutable query context.
public CultMeshQueryContext Build()
{
return new CultMeshQueryContext(_runtimeId, _routeHint);
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Typed derived-state query surface.
///
public sealed class CultMeshQuerySurface
{
private readonly Func> _execute;
private readonly Func>? _watch;
/// Creates a query surface.
public CultMeshQuerySurface(
string queryId,
Func> execute,
Func>? watch = null,
IEnumerable? sources = null,
CultMeshRouteHint? routeHint = null)
{
QueryId = RequireNonEmpty(queryId, nameof(queryId));
_execute = execute ?? throw new ArgumentNullException(nameof(execute));
_watch = watch;
Sources = sources?.ToArray() ?? Array.Empty();
RouteHint = routeHint ?? CultMeshRouteHint.Automatic;
}
/// Gets the semantic query id.
public string QueryId { get; }
/// Gets the typed state sources this query depends on, when known.
public IReadOnlyList Sources { get; }
/// Gets the preferred or observed route for the query.
public CultMeshRouteHint RouteHint { get; }
/// Executes the query once.
public Task ExecuteAsync(TParameters parameters, CultMeshQueryContext context)
{
if (context == null) throw new ArgumentNullException(nameof(context));
return _execute(parameters, context);
}
/// Executes the query once for one runtime using default context.
public Task ExecuteAsync(TParameters parameters, string runtimeId)
{
return ExecuteAsync(parameters, CultMeshQueryContext.ForRuntime(runtimeId));
}
/// Watches query results when the surface supports reactive execution.
public Observable Watch(TParameters parameters, CultMeshQueryContext context)
{
if (context == null) throw new ArgumentNullException(nameof(context));
if (_watch == null)
{
throw new NotSupportedException($"Query surface '{QueryId}' does not support reactive watches.");
}
return _watch(parameters, context);
}
/// Binds this query surface to a Verse context.
public CultMeshBoundQuerySurface Bind(CultMeshVerseContext context)
{
return new CultMeshBoundQuerySurface(context, this);
}
/// Binds this query surface to a Verse.
public CultMeshBoundQuerySurface Bind(CultMeshVerse verse)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Bind(verse.Context);
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Query surface pre-bound to a Verse so generated domain sugar can execute without context plumbing.
///
public sealed class CultMeshBoundQuerySurface
{
/// Creates a Verse-bound query surface.
public CultMeshBoundQuerySurface(
CultMeshVerseContext context,
CultMeshQuerySurface query)
{
Context = context ?? throw new ArgumentNullException(nameof(context));
Query = query ?? throw new ArgumentNullException(nameof(query));
}
/// Gets the bound Verse context.
public CultMeshVerseContext Context { get; }
/// Gets the underlying query surface.
public CultMeshQuerySurface Query { get; }
/// Gets the semantic query id.
public string QueryId => Query.QueryId;
/// Gets the typed state sources this query depends on, when known.
public IReadOnlyList Sources => Query.Sources;
/// Gets the preferred or observed route for the query.
public CultMeshRouteHint RouteHint => Query.RouteHint;
/// Executes the query through the bound Verse.
public Task ExecuteAsync(TParameters parameters)
{
return Query.ExecuteAsync(parameters, Context.QueryContext());
}
/// Watches query results through the bound Verse.
public Observable Watch(TParameters parameters)
{
return Query.Watch(parameters, Context.QueryContext());
}
}
///
/// Typed live view surface for composed client snapshots.
///
public sealed class CultMeshLiveFeed
{
private readonly Func> _snapshot;
private readonly Func>? _watch;
/// Creates a live feed surface.
public CultMeshLiveFeed(
string feedId,
Func> snapshot,
Func>? watch = null,
IEnumerable? sources = null,
CultMeshRouteHint? routeHint = null)
{
FeedId = RequireNonEmpty(feedId, nameof(feedId));
_snapshot = snapshot ?? throw new ArgumentNullException(nameof(snapshot));
_watch = watch;
Sources = sources?.ToArray() ?? Array.Empty();
RouteHint = routeHint ?? CultMeshRouteHint.Automatic;
}
/// Gets the semantic feed id.
public string FeedId { get; }
/// Gets the typed state sources this live feed depends on, when known.
public IReadOnlyList Sources { get; }
/// Gets the preferred or observed route for the feed.
public CultMeshRouteHint RouteHint { get; }
/// Reads one coherent feed snapshot.
public Task SnapshotAsync(TParameters parameters, CultMeshQueryContext context)
{
if (context == null) throw new ArgumentNullException(nameof(context));
return _snapshot(parameters, ResolveContext(context));
}
/// Reads one coherent feed snapshot for one runtime using default context.
public Task SnapshotAsync(TParameters parameters, string runtimeId)
{
return SnapshotAsync(parameters, CultMeshQueryContext.ForRuntime(runtimeId));
}
/// Watches coherent feed snapshots when the feed supports reactive execution.
public Observable Watch(TParameters parameters, CultMeshQueryContext context)
{
if (context == null) throw new ArgumentNullException(nameof(context));
if (_watch == null)
{
throw new NotSupportedException($"Live feed '{FeedId}' does not support reactive watches.");
}
return _watch(parameters, ResolveContext(context));
}
/// Binds this live feed to a Verse context.
public CultMeshBoundLiveFeed Bind(CultMeshVerseContext context)
{
return new CultMeshBoundLiveFeed(context, this);
}
/// Binds this live feed to a Verse.
public CultMeshBoundLiveFeed Bind(CultMeshVerse verse)
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return Bind(verse.Context);
}
private CultMeshQueryContext ResolveContext(CultMeshQueryContext context)
{
return context.RouteHint.Kind == CultMeshLocalityKind.Automatic &&
RouteHint.Kind != CultMeshLocalityKind.Automatic
? context.WithRoute(RouteHint)
: context;
}
private static string RequireNonEmpty(string value, string paramName)
{
return string.IsNullOrWhiteSpace(value)
? throw new ArgumentException("Value must be non-empty.", paramName)
: value;
}
}
///
/// Live feed pre-bound to a Verse so generated domain sugar can snapshot or watch without context plumbing.
///
public sealed class CultMeshBoundLiveFeed
{
/// Creates a Verse-bound live feed.
public CultMeshBoundLiveFeed(
CultMeshVerseContext context,
CultMeshLiveFeed feed)
{
Context = context ?? throw new ArgumentNullException(nameof(context));
Feed = feed ?? throw new ArgumentNullException(nameof(feed));
}
/// Gets the bound Verse context.
public CultMeshVerseContext Context { get; }
/// Gets the underlying live feed.
public CultMeshLiveFeed Feed { get; }
/// Gets the semantic feed id.
public string FeedId => Feed.FeedId;
/// Gets the typed state sources this live feed depends on, when known.
public IReadOnlyList Sources => Feed.Sources;
/// Gets the preferred or observed route for the feed.
public CultMeshRouteHint RouteHint => Feed.RouteHint;
/// Reads one coherent feed snapshot through the bound Verse.
public Task SnapshotAsync(TParameters parameters)
{
return Feed.SnapshotAsync(parameters, Context.QueryContext());
}
/// Watches coherent feed snapshots through the bound Verse.
public Observable Watch(TParameters parameters)
{
return Feed.Watch(parameters, Context.QueryContext());
}
}
///
/// Empty parameter object for document handles that expose one typed document as a live feed.
///
public readonly struct CultMeshDocumentQueryParameters
{
/// Gets the singleton empty document query.
public static CultMeshDocumentQueryParameters Empty { get; } = new();
}
///
/// Inspectable typed document handle exposed by CultMesh.
///
public interface ICultMeshDocumentHandle
{
/// Gets the CLR document type presented by this handle.
Type DocumentType { get; }
/// Gets the semantic document id.
string DocumentId { get; }
/// Gets the stable CultCache schema name.
string SchemaName { get; }
/// Gets the stable CultCache schema version.
string SchemaVersion { get; }
/// Gets the content-derived schema identifier.
string SchemaId { get; }
/// Gets the preferred or observed route for document access.
CultMeshRouteHint RouteHint { get; }
/// Gets typed state sources this document handle depends on, when known.
IReadOnlyList Sources { get; }
/// Gets whether this handle can replace the underlying document value.
bool CanReplace { get; }
/// Gets whether this handle can submit a client prediction for the underlying document value.
bool CanSubmitPrediction { get; }
/// Creates a same-schema alias presentation for another CLR document type.
CultMeshDocumentHandle AsSchemaAlias() where TAlias : class;
}
///
/// Typed reactive document handle with schema-aware alias conversion.
///
public sealed class CultMeshDocumentHandle : ICultMeshDocumentHandle
where TDocument : class
{
private static readonly CultDocumentDescriptor Descriptor =
CultDocumentRegistry.Shared.GetRequired();
private readonly CultMeshBoundLiveFeed _feed;
private readonly Func? _replace;
private readonly Func? _submitPrediction;
// Set only over a CultCache record, whose changes carry the cache's Sequence.
private readonly Func>>? _recordChanges;
private readonly Func<(TDocument Document, long Sequence)>? _sequencedLatest;
/// Creates a document handle from a Verse-bound live feed.
public CultMeshDocumentHandle(
CultMeshBoundLiveFeed feed,
Func? replace = null,
Func? submitPrediction = null)
{
_feed = feed ?? throw new ArgumentNullException(nameof(feed));
_replace = replace;
_submitPrediction = submitPrediction;
}
internal CultMeshDocumentHandle(
CultMeshBoundLiveFeed feed,
Func replace,
Func>> recordChanges,
Func<(TDocument Document, long Sequence)> sequencedLatest)
: this(feed, replace)
{
_recordChanges = recordChanges;
_sequencedLatest = sequencedLatest;
}
/// Gets the semantic document id.
public string DocumentId => _feed.FeedId;
/// Gets the CLR document type presented by this handle.
public Type DocumentType => typeof(TDocument);
/// Gets the stable CultCache schema name.
public string SchemaName => Descriptor.SchemaName;
/// Gets the stable CultCache schema version.
public string SchemaVersion => Descriptor.SchemaVersion;
/// Gets the content-derived schema identifier.
public string SchemaId => Descriptor.SchemaId;
/// Gets the preferred or observed route for document access.
public CultMeshRouteHint RouteHint => _feed.RouteHint;
/// Gets typed state sources this document handle depends on, when known.
public IReadOnlyList Sources => _feed.Sources;
/// Gets the bound Verse context.
public CultMeshVerseContext Context => _feed.Context;
/// Gets the underlying live feed used by this document handle.
public CultMeshLiveFeed Feed => _feed.Feed;
/// Gets whether this handle can replace the underlying document value.
public bool CanReplace => _replace != null;
/// Gets whether this handle can submit a client prediction for the underlying document value.
public bool CanSubmitPrediction => _submitPrediction != null;
/// Reads one coherent document snapshot.
public Task LatestAsync()
{
return _feed.SnapshotAsync(CultMeshDocumentQueryParameters.Empty);
}
/// Reads the latest document snapshot synchronously for host APIs that cannot be async.
public TDocument Latest()
{
return LatestAsync().ConfigureAwait(false).GetAwaiter().GetResult();
}
/// Watches coherent document snapshots.
public Observable Watch()
{
return _feed.Watch(CultMeshDocumentQueryParameters.Empty);
}
/// Subscribes to coherent document snapshots.
public IDisposable Watch(Action onNext)
{
if (onNext == null) throw new ArgumentNullException(nameof(onNext));
return Watch().Subscribe(onNext);
}
// Latest-value mirrors: a cache-backed handle passes each snapshot's Sequence; other sources pass null.
internal IDisposable WatchSequenced(Action onNext) => _recordChanges == null
? Watch(document => onNext(document, null))
: _recordChanges()
.Where(change => change.Document != null)
.Subscribe(change => onNext(change.Document!, change.Sequence));
// A mirror subscribes first, then adopts under its own gate. A cache-backed handle reads synchronously there, so
// its Sequence is at or above every change the mirror has applied and a later straggler at or below it drops.
internal async Task AdoptLatestAsync(object gate, Func adopt)
{
if (_sequencedLatest != null)
{
lock (gate)
{
var (document, sequence) = _sequencedLatest();
return adopt(document, sequence);
}
}
var latest = await LatestAsync().ConfigureAwait(false);
lock (gate)
return adopt(latest, null);
}
///
/// Projects one typed document field into a stable reactive state pointer. The expression is
/// structural metadata: only member access and constant collection indices are accepted.
///
public CultMeshStatePointer Field(Expression> field)
{
if (field == null) throw new ArgumentNullException(nameof(field));
var path = CultMeshFieldPath.Parse(field.Body, field.Parameters[0]);
var read = field.Compile();
return new CultMeshStatePointer(
DocumentId + "#" + path,
async context => read(await Feed
.SnapshotAsync(CultMeshDocumentQueryParameters.Empty, context)
.ConfigureAwait(false)),
context => Feed
.Watch(CultMeshDocumentQueryParameters.Empty, context)
.Select(read),
RouteHint,
Sources);
}
/// Replaces the underlying document when this handle is backed by mutable state.
public Task ReplaceAsync(TDocument value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
if (_replace == null)
{
throw new NotSupportedException($"Document handle '{DocumentId}' is read-only.");
}
return _replace(value);
}
/// Submits a locally predicted document value when this handle is backed by client-authoritative state.
public Task SubmitPredictionAsync(TDocument value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
if (_submitPrediction == null)
{
throw new NotSupportedException($"Document handle '{DocumentId}' does not accept client predictions.");
}
return _submitPrediction(value);
}
/// Gets an explicit authoritative replacement writer for this document.
public CultMeshDocumentWriter AuthoritativeWriter()
{
if (_replace == null)
throw new NotSupportedException($"Document handle '{DocumentId}' is read-only.");
return new CultMeshDocumentWriter(
this,
CultMeshDocumentWriteKind.AuthoritativeReplacement,
_replace);
}
/// Gets an explicit client prediction writer for this document.
public CultMeshDocumentWriter PredictionWriter()
{
if (_submitPrediction == null)
throw new NotSupportedException($"Document handle '{DocumentId}' does not accept client predictions.");
return new CultMeshDocumentWriter(
this,
CultMeshDocumentWriteKind.Prediction,
_submitPrediction);
}
/// Creates a read-only in-memory mirror updated by the document watch.
public async Task> ObserveAsync()
{
var observed = new CultMeshObservedDocument(this);
await observed.StartAsync().ConfigureAwait(false);
return observed;
}
/// Creates a read-only in-memory mirror synchronously.
public CultMeshObservedDocument Observe()
{
return ObserveAsync().ConfigureAwait(false).GetAwaiter().GetResult();
}
/// Creates a same-schema alias presentation for another CLR document type.
public CultMeshDocumentHandle AsSchemaAlias() where TAlias : class
{
var aliasDescriptor = CultDocumentRegistry.Shared.GetRequired();
if (!string.Equals(Descriptor.SchemaName, aliasDescriptor.SchemaName, StringComparison.Ordinal) ||
!string.Equals(Descriptor.SchemaVersion, aliasDescriptor.SchemaVersion, StringComparison.Ordinal))
{
throw new InvalidOperationException(
$"Document type {typeof(TAlias).FullName} uses schema '{aliasDescriptor.SchemaName}' " +
$"version '{aliasDescriptor.SchemaVersion}', but handle '{DocumentId}' exposes " +
$"schema '{SchemaName}' version '{SchemaVersion}'.");
}
var aliasFeed = new CultMeshLiveFeed(
DocumentId,
async (parameters, context) => ConvertDocument(
await Feed.SnapshotAsync(parameters, context).ConfigureAwait(false)),
(parameters, context) => Feed
.Watch(parameters, context)
.Select(ConvertDocument),
Sources,
RouteHint);
Func? replace = _replace == null
? null
: value => _replace(ConvertDocument(value));
Func? submitPrediction = _submitPrediction == null
? null
: value => _submitPrediction(ConvertDocument(value));
return new CultMeshDocumentHandle(aliasFeed.Bind(Context), replace, submitPrediction);
}
internal static TDocumentValue CloneDocument(TDocumentValue document)
where TDocumentValue : class
{
if (document == null) throw new ArgumentNullException(nameof(document));
var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TDocumentValue));
return (TDocumentValue)CultDocumentMessagePackSerialization.DeserializeUntyped(
typeof(TDocumentValue),
payload);
}
private static class CultMeshFieldPath
{
public static string Parse(Expression expression, ParameterExpression root)
{
var segments = new Stack();
while (expression != root)
{
if (expression is MemberExpression member)
{
segments.Push(member.Member.Name);
expression = member.Expression
?? throw Invalid();
continue;
}
if (expression is MethodCallExpression call &&
call.Method.Name == "get_Item" &&
call.Object != null &&
call.Arguments.Count == 1)
{
segments.Push("[" + ConstantIndex(call.Arguments[0]) + "]");
expression = call.Object;
continue;
}
if (expression is BinaryExpression binary && binary.NodeType == ExpressionType.ArrayIndex)
{
segments.Push("[" + ConstantIndex(binary.Right) + "]");
expression = binary.Left;
continue;
}
throw Invalid();
}
if (segments.Count == 0) throw Invalid();
var path = string.Empty;
foreach (var segment in segments)
path += segment.StartsWith("[", StringComparison.Ordinal) || path.Length == 0
? segment
: "." + segment;
return path;
}
private static string ConstantIndex(Expression expression)
{
if (expression is ConstantExpression constant && constant.Value != null)
return Convert.ToString(constant.Value, CultureInfo.InvariantCulture)!;
throw Invalid();
}
private static ArgumentException Invalid() => new(
"CultMesh field selectors may contain only member access and constant collection indices.",
"field");
}
internal static TTarget ConvertDocument(TSource document)
where TSource : class
where TTarget : class
{
if (document == null) throw new ArgumentNullException(nameof(document));
if (document is TTarget alreadyTyped)
{
return alreadyTyped;
}
var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TSource));
return (TTarget)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(TTarget), payload);
}
}
///
/// A read-only local mirror of a watched typed document. Observation never acquires a write
/// capability and never polls or serializes the document while idle.
///
public sealed class CultMeshObservedDocument : IDisposable
where TDocument : class
{
private readonly object _gate = new();
private IDisposable? _subscription;
private TDocument _current;
private long _appliedSequence;
private bool _disposed;
internal CultMeshObservedDocument(CultMeshDocumentHandle document)
{
Document = document ?? throw new ArgumentNullException(nameof(document));
_current = null!;
}
/// Gets the underlying read/watch handle.
public CultMeshDocumentHandle Document { get; }
///
/// Gets the current borrowed snapshot. Treat the object graph as read-only; local mutation
/// cannot publish and is replaced by the next canonical snapshot.
///
public TDocument Current
{
get
{
lock (_gate)
{
ThrowIfDisposed();
return _current;
}
}
}
internal async Task StartAsync()
{
_subscription = Document.WatchSequenced(ApplyCanonicalSnapshot);
try
{
await RefreshAsync().ConfigureAwait(false);
}
catch
{
Dispose();
throw;
}
}
/// Reads and adopts a fresh canonical snapshot.
public Task RefreshAsync()
{
ThrowIfDisposed();
return Document.AdoptLatestAsync(_gate, (latest, sequence) =>
{
ThrowIfDisposed();
_current = CultMeshDocumentHandle.CloneDocument(latest);
_appliedSequence = sequence ?? _appliedSequence;
return _current;
});
}
///
public void Dispose()
{
lock (_gate)
{
if (_disposed)
return;
_disposed = true;
}
_subscription?.Dispose();
}
private void ApplyCanonicalSnapshot(TDocument canonical, long? sequence)
{
if (canonical == null)
return;
var next = CultMeshDocumentHandle.CloneDocument(canonical);
lock (_gate)
{
if (_disposed || sequence <= _appliedSequence)
return;
_appliedSequence = sequence ?? _appliedSequence;
_current = next;
}
}
private void ThrowIfDisposed()
{
if (_disposed)
throw new ObjectDisposedException(nameof(CultMeshObservedDocument));
}
}
/// Names the one immutable authority path selected by a document writer.
public enum CultMeshDocumentWriteKind
{
/// The writer replaces canonical state through its owning provider.
AuthoritativeReplacement,
/// The writer submits a client prediction or candidate for validation.
Prediction
}
///
/// An explicit, immutable write capability for one typed document. A writer never switches
/// between authoritative replacement and prediction based on runtime availability.
///
public sealed class CultMeshDocumentWriter
where TDocument : class
{
private readonly Func _write;
internal CultMeshDocumentWriter(
CultMeshDocumentHandle document,
CultMeshDocumentWriteKind kind,
Func write)
{
Document = document ?? throw new ArgumentNullException(nameof(document));
Kind = kind;
_write = write ?? throw new ArgumentNullException(nameof(write));
}
/// Gets the document read/watch handle paired with this writer.
public CultMeshDocumentHandle Document { get; }
/// Gets the immutable authority path selected for this writer.
public CultMeshDocumentWriteKind Kind { get; }
/// Writes one value through this writer's named authority path.
public Task WriteAsync(TDocument value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
return _write(value);
}
/// Reads, transforms, and writes through this writer's named authority path.
public async Task UpdateAsync(Func update)
{
if (update == null) throw new ArgumentNullException(nameof(update));
var next = update(await Document.LatestAsync().ConfigureAwait(false));
await WriteAsync(next).ConfigureAwait(false);
return next;
}
/// Reads, transforms, and writes through this writer's named authority path.
public async Task UpdateAsync(Func> update)
{
if (update == null) throw new ArgumentNullException(nameof(update));
var next = await update(await Document.LatestAsync().ConfigureAwait(false)).ConfigureAwait(false);
await WriteAsync(next).ConfigureAwait(false);
return next;
}
/// Creates an explicitly writable reactive mirror.
public async Task> ReactiveAsync(
CultMeshReactiveDocumentOptions? options = null)
{
var reactive = new CultMeshReactiveDocument(this, options);
await reactive.StartAsync().ConfigureAwait(false);
return reactive;
}
/// Creates an explicitly writable reactive mirror synchronously.
public CultMeshReactiveDocument Reactive(CultMeshReactiveDocumentOptions? options = null)
{
return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult();
}
}
///
/// Configures a managed CultMesh reactive document mirror.
///
public sealed class CultMeshReactiveDocumentOptions
{
///
/// Gets or sets the write coalescing window. The default approximates a frame boundary for non-Unity hosts.
///
public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16);
///
/// Gets or sets the clock that owns delayed flush scheduling. Tests and deterministic hosts may
/// replace the system clock without introducing a second scheduler abstraction.
///
public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance;
/// Gets or sets whether canonical snapshots replace dirty local predictions immediately.
public bool ReplaceDirtyCurrentOnCanonicalSnapshot { get; set; }
}
///
/// Captures a canonical snapshot that arrived while the local reactive document had an outstanding prediction.
///
public sealed class CultMeshReactiveDocumentReconciliation
where TDocument : class
{
internal CultMeshReactiveDocumentReconciliation(
TDocument canonical,
TDocument predicted,
IReadOnlyDictionary delta,
int version,
DateTimeOffset receivedAt)
{
Canonical = canonical;
Predicted = predicted;
Delta = delta ?? throw new ArgumentNullException(nameof(delta));
Version = version;
ReceivedAt = receivedAt;
}
/// Gets the monotonically increasing reconciliation version for this reactive document.
public int Version { get; }
/// Gets the authoritative canonical document snapshot.
public TDocument Canonical { get; }
/// Gets the locally predicted document snapshot that was active when the canonical value arrived.
public TDocument Predicted { get; }
///
/// Gets the predicted-vs-canonical member delta. Numeric members store predicted minus canonical;
/// non-numeric members store the predicted value.
///
public IReadOnlyDictionary Delta { get; }
/// Gets when the canonical snapshot was received.
public DateTimeOffset ReceivedAt { get; }
}
///
/// Managed typed document mirror that coalesces explicit local edits through one selected writer.
///
public sealed class CultMeshReactiveDocument : IDisposable
where TDocument : class
{
private static readonly CultMeshComparableMember[] ComparableMembers =
GetComparableMembers(typeof(TDocument)).ToArray();
private readonly CultMeshDocumentWriter _writer;
private readonly CultMeshDocumentHandle _document;
private readonly CultMeshReactiveDocumentOptions _options;
private readonly object _gate = new();
private IDisposable? _subscription;
private CancellationTokenSource? _flushDelayCancellation;
private TDocument _current;
private bool _dirty;
private bool _flushQueued;
private bool _flushing;
private bool _disposed;
private long _appliedSequence;
private int _reconciliationVersion;
internal CultMeshReactiveDocument(
CultMeshDocumentWriter writer,
CultMeshReactiveDocumentOptions? options)
{
_writer = writer ?? throw new ArgumentNullException(nameof(writer));
_document = writer.Document;
_current = null!;
_options = options ?? new CultMeshReactiveDocumentOptions();
}
/// Gets the immutable write capability selected for this mirror.
public CultMeshDocumentWriter Writer => _writer;
/// Gets the underlying document handle.
public CultMeshDocumentHandle Document => _document;
///
/// Gets a cloned local snapshot. Mutating the returned object cannot mutate or publish the mirror;
/// use or for edits.
///
public TDocument Snapshot
{
get
{
lock (_gate)
{
ThrowIfDisposed();
return CultMeshDocumentHandle.CloneDocument(_current);
}
}
}
/// Gets whether local edits are waiting to be sent.
public bool IsDirty
{
get
{
lock (_gate)
return _dirty;
}
}
/// Gets the most recent reconciliation snapshot, when a canonical value arrived during local prediction.
public CultMeshReactiveDocumentReconciliation? Reconciliation { get; private set; }
internal async Task StartAsync()
{
_subscription = _document.WatchSequenced(ApplyCanonicalSnapshot);
try
{
await RefreshAsync().ConfigureAwait(false);
}
catch
{
Dispose();
throw;
}
}
/// Mutates the current value and schedules a coalesced prediction or replacement.
public TDocument Update(Action update)
{
if (update == null) throw new ArgumentNullException(nameof(update));
ThrowIfDisposed();
lock (_gate)
{
update(_current);
_dirty = true;
ScheduleFlushLocked();
return CultMeshDocumentHandle.CloneDocument(_current);
}
}
/// Replaces the current local value and schedules a coalesced prediction or replacement.
public TDocument ReplaceLocal(TDocument value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
ThrowIfDisposed();
lock (_gate)
{
_current = CultMeshDocumentHandle.CloneDocument(value);
_dirty = true;
ScheduleFlushLocked();
return CultMeshDocumentHandle.CloneDocument(_current);
}
}
/// Reads a fresh canonical snapshot and adopts it as the current value.
public Task RefreshAsync()
{
ThrowIfDisposed();
return _document.AdoptLatestAsync(_gate, (latest, sequence) =>
{
_current = CultMeshDocumentHandle.CloneDocument(latest);
_appliedSequence = sequence ?? _appliedSequence;
_dirty = false;
Reconciliation = null;
return CultMeshDocumentHandle.CloneDocument(_current);
});
}
/// Immediately sends the latest local dirty value, if any, through the document authority shape.
public async Task FlushAsync()
{
TDocument predicted;
lock (_gate)
{
ThrowIfDisposed();
if (!_dirty)
return;
if (_flushing)
{
_flushQueued = true;
return;
}
CancelScheduledFlushLocked();
_flushing = true;
_dirty = false;
predicted = CultMeshDocumentHandle.CloneDocument(_current);
}
try
{
await _writer.WriteAsync(predicted).ConfigureAwait(false);
}
finally
{
var shouldFlushAgain = false;
lock (_gate)
{
_flushing = false;
shouldFlushAgain = _flushQueued || _dirty;
_flushQueued = false;
}
if (shouldFlushAgain)
await FlushAsync().ConfigureAwait(false);
}
}
/// Clears the most recent reconciliation metadata.
public void ClearReconciliation()
{
ThrowIfDisposed();
Reconciliation = null;
}
///
public void Dispose()
{
lock (_gate)
{
if (_disposed)
return;
_disposed = true;
CancelScheduledFlushLocked();
}
_subscription?.Dispose();
}
private void ApplyCanonicalSnapshot(TDocument canonical, long? sequence)
{
if (canonical == null)
return;
lock (_gate)
{
if (_disposed || sequence <= _appliedSequence)
return;
_appliedSequence = sequence ?? _appliedSequence;
TDocument? replacement = null;
if (_dirty || _flushing)
{
var delta = CreateReconciliationDelta(_current, canonical);
if (delta.Count == 0)
{
Reconciliation = null;
if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot)
return;
}
else
{
var nextCanonical = CultMeshDocumentHandle.CloneDocument(canonical);
var predicted = CultMeshDocumentHandle.CloneDocument(_current);
replacement = nextCanonical;
Reconciliation = new CultMeshReactiveDocumentReconciliation(
nextCanonical,
predicted,
delta,
++_reconciliationVersion,
DateTimeOffset.UtcNow);
if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot)
return;
}
}
_current = replacement ?? CultMeshDocumentHandle.CloneDocument(canonical);
Reconciliation = null;
}
}
private void ScheduleFlushLocked()
{
if (_options.FlushDelay <= TimeSpan.Zero)
{
_ = Task.Run(FlushAsync);
return;
}
CancelScheduledFlushLocked();
var cancellation = new CancellationTokenSource();
_flushDelayCancellation = cancellation;
_ = FlushAfterDelayAsync(cancellation);
}
private async Task FlushAfterDelayAsync(CancellationTokenSource cancellation)
{
try
{
await _options.Clock.DelayAsync(_options.FlushDelay, cancellation.Token).ConfigureAwait(false);
lock (_gate)
{
if (_disposed || cancellation.IsCancellationRequested)
return;
}
await FlushAsync().ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellation.IsCancellationRequested)
{
}
finally
{
lock (_gate)
{
if (ReferenceEquals(_flushDelayCancellation, cancellation))
_flushDelayCancellation = null;
}
cancellation.Dispose();
}
}
private void CancelScheduledFlushLocked()
{
var cancellation = _flushDelayCancellation;
_flushDelayCancellation = null;
if (cancellation == null)
return;
cancellation.Cancel();
}
private static IReadOnlyDictionary CreateReconciliationDelta(
TDocument predicted,
TDocument canonical)
{
var delta = new Dictionary(StringComparer.Ordinal);
foreach (var member in ComparableMembers)
{
var predictedValue = member.GetValue(predicted);
var canonicalValue = member.GetValue(canonical);
if (CultMeshValuesEqual(predictedValue, canonicalValue))
continue;
delta[member.Name] = TryCreateNumericDelta(predictedValue, canonicalValue, out var numericDelta)
? numericDelta
: predictedValue;
}
return delta;
}
private static IEnumerable GetComparableMembers(Type documentType)
{
foreach (var property in documentType.GetProperties(BindingFlags.Instance | BindingFlags.Public))
{
if (property.GetMethod == null || property.GetIndexParameters().Length != 0)
continue;
yield return new CultMeshComparableMember(property.Name, property.GetValue);
}
foreach (var field in documentType.GetFields(BindingFlags.Instance | BindingFlags.Public))
{
yield return new CultMeshComparableMember(field.Name, field.GetValue);
}
}
private static bool TryCreateNumericDelta(object? predicted, object? canonical, out object? delta)
{
delta = null;
if (predicted == null || canonical == null)
return false;
var predictedType = Nullable.GetUnderlyingType(predicted.GetType()) ?? predicted.GetType();
var canonicalType = Nullable.GetUnderlyingType(canonical.GetType()) ?? canonical.GetType();
if (!IsNumericType(predictedType) || !IsNumericType(canonicalType))
return false;
delta = Convert.ToDouble(predicted, CultureInfo.InvariantCulture) -
Convert.ToDouble(canonical, CultureInfo.InvariantCulture);
return true;
}
private static bool IsNumericType(Type type)
{
if (type.IsEnum)
return false;
return Type.GetTypeCode(type) switch
{
TypeCode.Byte or
TypeCode.SByte or
TypeCode.UInt16 or
TypeCode.UInt32 or
TypeCode.UInt64 or
TypeCode.Int16 or
TypeCode.Int32 or
TypeCode.Int64 or
TypeCode.Decimal or
TypeCode.Double or
TypeCode.Single => true,
_ => false
};
}
private static bool CultMeshValuesEqual(object? left, object? right)
{
if (ReferenceEquals(left, right))
return true;
if (left == null || right == null)
return false;
if (left.Equals(right))
return true;
if (left is string || right is string)
return false;
if (left is IEnumerable leftEnumerable && right is IEnumerable rightEnumerable)
return leftEnumerable.Cast().SequenceEqual(rightEnumerable.Cast());
return false;
}
private void ThrowIfDisposed()
{
if (_disposed)
throw new ObjectDisposedException(nameof(CultMeshReactiveDocument));
}
private readonly struct CultMeshComparableMember
{
public CultMeshComparableMember(string name, Func getValue)
{
Name = name;
GetValue = getValue;
}
public string Name { get; }
public Func GetValue { get; }
}
}
///
/// Schema-aware lookup table for typed CultMesh document handles.
///
public sealed class CultMeshDocumentCatalog
{
private readonly IReadOnlyList _documents;
private readonly IReadOnlyDictionary _documentsByType;
private readonly IReadOnlyDictionary _documentsBySchema;
/// Creates a document catalog from known handles.
public CultMeshDocumentCatalog(IEnumerable documents)
{
if (documents == null) throw new ArgumentNullException(nameof(documents));
var documentList = documents
.Where(document => document != null)
.ToArray();
var byType = new Dictionary();
var bySchema = new Dictionary(StringComparer.Ordinal);
foreach (var document in documentList)
{
byType[document.DocumentType] = document;
if (!string.IsNullOrWhiteSpace(document.SchemaId))
bySchema[document.SchemaId] = document;
if (!string.IsNullOrWhiteSpace(document.SchemaVersion))
bySchema[document.SchemaVersion] = document;
if (!string.IsNullOrWhiteSpace(document.SchemaName))
bySchema[document.SchemaName] = document;
}
_documents = documentList;
_documentsByType = byType;
_documentsBySchema = bySchema;
}
/// Gets the handles in this catalog.
public IReadOnlyList Documents => _documents;
/// Looks up a document handle by schema id, shared schema name, or schema version.
public bool TryGetDocumentBySchema(
string schema,
out ICultMeshDocumentHandle document)
{
if (!string.IsNullOrWhiteSpace(schema) &&
_documentsBySchema.TryGetValue(schema, out document!))
{
return true;
}
document = null!;
return false;
}
/// Looks up a document handle by schema id, shared schema name, or schema version.
public ICultMeshDocumentHandle DocumentBySchema(string schema)
{
if (TryGetDocumentBySchema(schema, out var document))
return document;
throw new NotSupportedException(
$"CultMesh document catalog does not expose a document for schema '{schema}'.");
}
/// Looks up a typed document handle by CLR type or same-schema alias.
public bool TryGetDocument(
out CultMeshDocumentHandle document)
where TDocument : class
{
if (_documentsByType.TryGetValue(typeof(TDocument), out var untypedDocument) &&
untypedDocument is CultMeshDocumentHandle typedDocument)
{
document = typedDocument;
return true;
}
var descriptor = CultDocumentRegistry.Shared.GetRequired();
if (_documentsBySchema.TryGetValue(descriptor.SchemaVersion, out var schemaDocument) ||
_documentsBySchema.TryGetValue(descriptor.SchemaName, out schemaDocument!))
{
if (schemaDocument is CultMeshDocumentHandle schemaTypedDocument)
{
document = schemaTypedDocument;
return true;
}
document = schemaDocument.AsSchemaAlias();
return true;
}
document = null!;
return false;
}
/// Looks up a typed document handle by CLR type or same-schema alias.
public CultMeshDocumentHandle Document()
where TDocument : class
{
if (TryGetDocument(out var document))
return document;
throw new NotSupportedException(
$"CultMesh document catalog does not expose a document for {typeof(TDocument).FullName}.");
}
/// Reads one typed document snapshot by CLR type or same-schema alias.
public Task LatestAsync()
where TDocument : class
{
return Document().LatestAsync();
}
/// Reads one typed document synchronously by CLR type or same-schema alias.
public TDocument Latest()
where TDocument : class
{
return Document().Latest();
}
/// Gets whether one typed document can be replaced by CLR type or same-schema alias.
public bool CanReplace()
where TDocument : class
{
return TryGetDocument(out var document) && document.CanReplace;
}
/// Replaces one typed document by CLR type or same-schema alias.
public Task ReplaceAsync(TDocument value)
where TDocument : class
{
return Document().ReplaceAsync(value);
}
/// Gets whether one typed document can submit client predictions by CLR type or same-schema alias.
public bool CanSubmitPrediction()
where TDocument : class
{
return TryGetDocument(out var document) && document.CanSubmitPrediction;
}
/// Submits a client prediction for one typed document by CLR type or same-schema alias.
public Task SubmitPredictionAsync(TDocument value)
where TDocument : class
{
return Document().SubmitPredictionAsync(value);
}
/// Gets an explicit authoritative writer by CLR type or same-schema alias.
public CultMeshDocumentWriter AuthoritativeWriter()
where TDocument : class
{
return Document().AuthoritativeWriter();
}
/// Gets an explicit prediction writer by CLR type or same-schema alias.
public CultMeshDocumentWriter PredictionWriter()
where TDocument : class
{
return Document().PredictionWriter();
}
/// Watches one typed document by CLR type or same-schema alias.
public Observable Watch()
where TDocument : class
{
return Document().Watch();
}
/// Subscribes to one typed document by CLR type or same-schema alias.
public IDisposable Watch(Action onNext)
where TDocument : class
{
return Document().Watch(onNext);
}
}
///
/// Inspectable typed collection handle exposed by CultMesh.
///
public interface ICultMeshCollectionHandle
{
/// Gets the CLR document type presented by this collection.
Type DocumentType { get; }
/// Gets the semantic collection id.
string CollectionId { get; }
/// Gets the stable CultCache schema name.
string SchemaName { get; }
/// Gets the stable CultCache schema version.
string SchemaVersion { get; }
/// Gets the content-derived schema identifier.
string SchemaId { get; }
/// Gets the preferred or observed route for collection access.
CultMeshRouteHint RouteHint { get; }
/// Gets typed state sources this collection handle depends on, when known.
IReadOnlyList Sources { get; }
/// Creates a same-schema alias presentation for another CLR document type.
CultMeshCollectionHandle AsSchemaAlias() where TAlias : class;
}
///
/// Schema-aware lookup table for typed CultMesh collection handles.
///
public sealed class CultMeshCollectionCatalog
{
private readonly IReadOnlyList _collections;
private readonly IReadOnlyDictionary _collectionsByType;
private readonly IReadOnlyDictionary _collectionsBySchema;
/// Creates a collection catalog from known handles.
public CultMeshCollectionCatalog(IEnumerable collections)
{
if (collections == null) throw new ArgumentNullException(nameof(collections));
var collectionList = collections
.Where(collection => collection != null)
.ToArray();
var byType = new Dictionary();
var bySchema = new Dictionary(StringComparer.Ordinal);
foreach (var collection in collectionList)
{
byType[collection.DocumentType] = collection;
if (!string.IsNullOrWhiteSpace(collection.SchemaId))
bySchema[collection.SchemaId] = collection;
if (!string.IsNullOrWhiteSpace(collection.SchemaVersion))
bySchema[collection.SchemaVersion] = collection;
if (!string.IsNullOrWhiteSpace(collection.SchemaName))
bySchema[collection.SchemaName] = collection;
}
_collections = collectionList;
_collectionsByType = byType;
_collectionsBySchema = bySchema;
}
/// Gets the handles in this catalog.
public IReadOnlyList Collections => _collections;
/// Looks up a collection handle by schema id, shared schema name, or schema version.
public bool TryGetCollectionBySchema(
string schema,
out ICultMeshCollectionHandle collection)
{
if (!string.IsNullOrWhiteSpace(schema) &&
_collectionsBySchema.TryGetValue(schema, out collection!))
{
return true;
}
collection = null!;
return false;
}
/// Looks up a collection handle by schema id, shared schema name, or schema version.
public ICultMeshCollectionHandle CollectionBySchema(string schema)
{
if (TryGetCollectionBySchema(schema, out var collection))
return collection;
throw new NotSupportedException(
$"CultMesh collection catalog does not expose a collection for schema '{schema}'.");
}
/// Looks up a typed collection handle by CLR type or same-schema alias.
public bool TryGetCollection(
out CultMeshCollectionHandle collection)
where TDocument : class
{
if (_collectionsByType.TryGetValue(typeof(TDocument), out var untypedCollection) &&
untypedCollection is CultMeshCollectionHandle typedCollection)
{
collection = typedCollection;
return true;
}
var descriptor = CultDocumentRegistry.Shared.GetRequired();
if (_collectionsBySchema.TryGetValue(descriptor.SchemaId, out var schemaCollection) ||
_collectionsBySchema.TryGetValue(descriptor.SchemaVersion, out schemaCollection!) ||
_collectionsBySchema.TryGetValue(descriptor.SchemaName, out schemaCollection!))
{
if (schemaCollection is CultMeshCollectionHandle schemaTypedCollection)
{
collection = schemaTypedCollection;
return true;
}
collection = schemaCollection.AsSchemaAlias();
return true;
}
collection = null!;
return false;
}
/// Looks up a typed collection handle by CLR type or same-schema alias.
public CultMeshCollectionHandle Collection()
where TDocument : class
{
if (TryGetCollection(out var collection))
return collection;
throw new NotSupportedException(
$"CultMesh collection catalog does not expose a collection for {typeof(TDocument).FullName}.");
}
/// Reads one typed collection snapshot by CLR type or same-schema alias.
public Task> LatestAsync()
where TDocument : class
{
return Collection().LatestAsync();
}
/// Watches one typed collection by CLR type or same-schema alias.
public Observable> WatchChanges()
where TDocument : class
{
return Collection().WatchChanges();
}
/// Subscribes to one typed collection by CLR type or same-schema alias.
public IDisposable WatchChanges(Action> onNext)
where TDocument : class
{
return Collection().WatchChanges(onNext);
}
}
///
/// Describes how a CultMesh collection document changed.
///
public enum CultMeshCollectionChangeKind
{
/// A document was added.
Added,
/// A document was updated.
Updated,
/// A document was removed.
Removed,
/// A local prediction was published before authority accepted it.
Predicted,
/// A predicted document was reconciled with authority.
Reconciled,
/// A document was accepted through schema migration.
SchemaMigrated,
/// A document change was rejected.
Rejected
}
///
/// One typed document change observed through a CultMesh collection handle.
///
public sealed class CultMeshCollectionChange
where TDocument : class
{
/// Creates a collection change.
public CultMeshCollectionChange(
CultMeshCollectionChangeKind kind,
CultRecordKey key,
string schemaId,
TDocument? document,
TDocument? previousDocument,
string? message = null)
{
Kind = kind;
Key = key;
SchemaId = schemaId ?? "";
Document = document;
PreviousDocument = previousDocument;
Message = message;
}
/// Gets the change kind.
public CultMeshCollectionChangeKind Kind { get; }
/// Gets the changed record key.
public CultRecordKey Key { get; }
/// Gets the schema id of the changed document.
public string SchemaId { get; }
/// Gets the current document, when present.
public TDocument? Document { get; }
/// Gets the previous document, when present.
public TDocument? PreviousDocument { get; }
/// Gets an optional rejection or diagnostic message.
public string? Message { get; }
}
///
/// Typed reactive collection handle with schema-aware alias conversion.
///
public sealed class CultMeshCollectionHandle : ICultMeshCollectionHandle
where TDocument : class
{
private static readonly CultDocumentDescriptor Descriptor =
CultDocumentRegistry.Shared.GetRequired