using System; using System.Collections.Generic; using System.Globalization; using System.Linq; using System.Threading.Tasks; using GameCult.Caching; using GameCult.Caching.MessagePack; using MessagePack; namespace GameCult.Networking { /// /// Optional metadata applied when creating CultNet document messages. /// public sealed class CultNetDocumentMessageOptions { /// /// Gets or sets an explicit storage timestamp. /// public string? StoredAt { get; set; } /// /// Gets or sets the runtime that produced the document. /// public string? SourceRuntimeId { get; set; } /// /// Gets or sets the agent that produced the document. /// public string? SourceAgentId { get; set; } /// /// Gets or sets the role that produced the document. /// public string? SourceRole { get; set; } /// /// Gets or sets optional document tags. /// public string[]? Tags { get; set; } } /// /// Binds a CultCache document type to CultNet payload serialization. /// public sealed class CultNetDocumentBinding { internal CultNetDocumentBinding( Type documentType, string schemaId, Func payloadSerializer, Func payloadDeserializer) { DocumentType = documentType; SchemaId = schemaId; PayloadSerializer = payloadSerializer; PayloadDeserializer = payloadDeserializer; } /// /// Gets the bound document type. /// public Type DocumentType { get; } /// /// Gets the bound schema identifier. /// public string SchemaId { get; } /// /// Gets the payload serializer. /// public Func PayloadSerializer { get; } /// /// Gets the payload deserializer. /// public Func PayloadDeserializer { get; } /// /// Creates a document binding for a typed CultCache document. /// public static CultNetDocumentBinding ForDocument( CultDocumentRegistry? registry = null, string? schemaId = null, Func? payloadSerializer = null, Func? payloadDeserializer = null) where T : class { var documents = registry ?? CultDocumentRegistry.Shared; var descriptor = documents.GetRequired(); return new CultNetDocumentBinding( typeof(T), string.IsNullOrWhiteSpace(schemaId) ? descriptor.SchemaId : schemaId!, document => { var typed = (T)document; return payloadSerializer != null ? payloadSerializer(typed) : CultDocumentMessagePackSerialization.SerializeUntyped(typed, typeof(T), documents); }, payload => payloadDeserializer != null ? payloadDeserializer(payload) : (T)CultDocumentMessagePackSerialization.DeserializeUntyped(typeof(T), payload, documents)); } /// /// Creates a document binding for a CultCache document type discovered at runtime. /// public static CultNetDocumentBinding ForDocument( Type documentType, CultDocumentRegistry? registry = null, string? schemaId = null, Func? payloadSerializer = null, Func? payloadDeserializer = null) { if (documentType == null) { throw new ArgumentNullException(nameof(documentType)); } var documents = registry ?? CultDocumentRegistry.Shared; var descriptor = documents.GetRequired(documentType); return new CultNetDocumentBinding( documentType, string.IsNullOrWhiteSpace(schemaId) ? descriptor.SchemaId : schemaId!, document => { if (!documentType.IsInstanceOfType(document)) { throw new InvalidOperationException( $"Document payload type {document.GetType().FullName} is not assignable to {documentType.FullName}."); } return payloadSerializer != null ? payloadSerializer(document) : CultDocumentMessagePackSerialization.SerializeUntyped(document, documentType, documents); }, payload => payloadDeserializer != null ? payloadDeserializer(payload) : CultDocumentMessagePackSerialization.DeserializeUntyped(documentType, payload, documents)); } } /// /// Creates and applies CultNet messages that replicate CultCache documents. /// public sealed class CultNetDocumentRegistry { private readonly CultDocumentRegistry _documents; private readonly Dictionary _bindingsBySchemaId = new(StringComparer.Ordinal); private readonly Dictionary _bindingsByType = new(); /// /// Creates a document registry with optional bindings. /// public CultNetDocumentRegistry( CultDocumentRegistry? documents = null, IEnumerable? bindings = null) { _documents = documents ?? CultDocumentRegistry.Shared; if (bindings == null) { return; } foreach (var binding in bindings) { Register(binding); } } /// /// Registers a document binding. /// public CultNetDocumentRegistry Register(CultNetDocumentBinding binding) { _bindingsBySchemaId[binding.SchemaId] = binding; _bindingsByType[binding.DocumentType] = binding; return this; } /// /// Gets a binding by schema identifier. /// public CultNetDocumentBinding? GetBySchemaId(string schemaId) { return _bindingsBySchemaId.TryGetValue(schemaId, out var binding) ? binding : null; } /// /// Gets a binding by CLR document type. /// public CultNetDocumentBinding? GetByDocumentType(Type documentType) { return _bindingsByType.TryGetValue(documentType, out var binding) ? binding : null; } /// /// Creates a document delete message. /// public CultNetDocumentDeleteMessage CreateDocumentDeleteMessage( string messageId, string schemaId, string recordKey) { return new CultNetDocumentDeleteMessage { MessageId = RequireNonEmpty(messageId, nameof(messageId)), SchemaId = RequireNonEmpty(schemaId, nameof(schemaId)), RecordKey = RequireNonEmpty(recordKey, nameof(recordKey)) }; } /// /// Creates a raw document put message for a typed document. /// public CultNetDocumentPutRawMessage CreateRawDocumentPutMessage( string messageId, CultRecordHandle handle, T document, CultNetDocumentMessageOptions? options = null) where T : class { var descriptor = _documents.GetRequired(); var binding = GetByDocumentType(typeof(T)) ?? CultNetDocumentBinding.ForDocument(_documents); var storedAt = ResolveStoredAt(options); return new CultNetDocumentPutRawMessage { MessageId = RequireNonEmpty(messageId, nameof(messageId)), Document = new CultNetRawDocumentRecord { SchemaId = binding.SchemaId, SchemaName = descriptor.SchemaName, SchemaVersion = descriptor.SchemaVersion, SchemaContentHash = descriptor.ContentHash, RecordKey = handle.Key.Value, StoredAt = storedAt, PayloadEncoding = "messagepack", Payload = binding.PayloadSerializer(document), SourceRuntimeId = options?.SourceRuntimeId, SourceAgentId = options?.SourceAgentId, SourceRole = options?.SourceRole, Tags = options?.Tags } }; } /// /// Creates a snapshot request message. /// public CultNetSnapshotRequestMessage CreateSnapshotRequest( string messageId, IEnumerable? schemaIds = null, IEnumerable? recordKeys = null) { return new CultNetSnapshotRequestMessage { MessageId = RequireNonEmpty(messageId, nameof(messageId)), SchemaIds = schemaIds?.ToArray(), RecordKeys = recordKeys?.ToArray() }; } /// /// Creates a typed-selection snapshot request message (docs/cultnet-selection-cut.md). /// public CultNetSnapshotRequestV1Message CreateSnapshotRequestV1(string messageId, CultNetSelection selection) { if (selection == null) throw new ArgumentNullException(nameof(selection)); return new CultNetSnapshotRequestV1Message { MessageId = RequireNonEmpty(messageId, nameof(messageId)), Selection = selection }; } /// /// Creates a raw snapshot response from the cache. v0 has no selector engine of its own /// (docs/cultnet-selection-cut.md, D3/D4): the filter lowers into a /// and is answered by in one evaluation (R-G), so a v0 /// caller sees its whole matching set in one message without re-filtering the cache per page. /// public CultNetSnapshotResponseRawMessage CreateRawSnapshotResponse( CultCache cache, string messageId, CultNetSnapshotRequestMessage? filter = null, CultNetDocumentMessageOptions? options = null) { if (cache == null) throw new ArgumentNullException(nameof(cache)); var lowSchemas = CultNetV0SelectionLowering.Lower(filter?.SchemaIds); var lowKeys = CultNetV0SelectionLowering.Lower(filter?.RecordKeys); // R-R: an explicit empty schemas or keys list is v0's own "answer nothing" - it must not // reach the v1 door, which refuses an empty selection.schemas/keys outright // (CultNetSelectionValidation.ValidateShape). v0 and v1 disagree about what an empty list // means, so v0 decides this before the v1 evaluator ever sees the selection. if (lowSchemas is { Length: 0 } || lowKeys is { Length: 0 }) { return new CultNetSnapshotResponseRawMessage { MessageId = RequireNonEmpty(messageId, nameof(messageId)), Documents = Array.Empty() }; } var selection = new CultNetSelection { Schemas = lowSchemas, Keys = lowKeys, Projection = CultNetSelectionProjections.Document }; // R-R: v0's pre-cut CreateRawSnapshotResponse (b3d9cf7) iterated the requested record keys // in the order the caller sent them (a HashSet built from the array, enumerated in // insertion order). The evaluator's own tiebreak (schema then key) only applies when the // caller did not name keys. Func ordinalOf; if (lowKeys is { Length: > 0 }) { var order = new Dictionary(StringComparer.Ordinal); for (var i = 0; i < lowKeys.Length; i++) order.TryAdd(lowKeys[i], i); ordinalOf = (_, key) => order.TryGetValue(key.Value, out var index) ? index : lowKeys.Length; } else { ordinalOf = static (_, _) => 0; } var page = SelectAll(cache, selection, ordinalOf, asOf: 0, options); return new CultNetSnapshotResponseRawMessage { MessageId = RequireNonEmpty(messageId, nameof(messageId)), Documents = page.Documents ?? Array.Empty() }; } /// /// Evaluates a typed selection against the cache and returns its page /// (docs/cultnet-selection-cut.md, section 2/6). supplies each /// row's ordinal (the reference: ); /// is the snapshot the page is exact for when the selection is not shard-scoped, or the fallback /// used only when is null or the selection matches no rows (S-9). /// public CultNetSnapshotResponseRawV1Message CreateSelectionResponse( CultCache cache, string messageId, CultNetSelection selection, Func ordinalOf, ulong asOf, CultNetDocumentMessageOptions? options = null, Func? shardIdOf = null, CultNetSelectionCursorKey? cursorKey = null, Func? asOfForShard = null) { if (cache == null) throw new ArgumentNullException(nameof(cache)); if (selection == null) throw new ArgumentNullException(nameof(selection)); if (ordinalOf == null) throw new ArgumentNullException(nameof(ordinalOf)); var page = SelectPage(cache, selection, ordinalOf, asOf, options, shardIdOf: shardIdOf, cursorKey: cursorKey, asOfForShard: asOfForShard); return new CultNetSnapshotResponseRawV1Message { MessageId = RequireNonEmpty(messageId, nameof(messageId)), Matched = page.Matched, AsOf = page.AsOf, Next = page.Next, Headers = page.Headers, Documents = page.Documents, Edges = page.Edges }; } /// /// Evaluates a selection over every row the cache holds and projects the matched page to wire /// records. The one evaluator () owns matching, order, /// the hop and the cursor; this owns turning its result into s. /// internal CultNetSelectionPage SelectPage( CultCache cache, CultNetSelection selection, Func ordinalOf, ulong asOf, CultNetDocumentMessageOptions? options, Func? rowFilter = null, Func? shardIdOf = null, CultNetSelectionCursorKey? cursorKey = null, Func? asOfForShard = null) { selection = ExpandSchemaBindingAliases(selection); // R-F: CultNetSelectionEvaluator.Select validates first, every time - there is no separate // validate flag here any more, because a public evaluation path that can skip the door is // exactly what R-F closes. var stored = cache.AllStoredDocuments; if (rowFilter != null) stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key)); var rows = stored .Select(entry => new CultNetSelectionEvaluator.Row( entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt)) .ToArray(); var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection); // R-Q: a page's asOf is exact only when every matched row is committed through one shard's // log - a caller that knows about shards (CultNetDatabaseServer) supplies shardIdOf; one // that does not (the raw v0/shard-scoped rowFilter paths, which are already bounded to a // single shard by construction) passes null and skips the check rather than pay for it. if (shardIdOf != null) { var shardId = EnsureSingleShard(full.Ordered, shardIdOf); // S-9: asOf is the watermark of the shard the selection's matched rows actually come // from, not the database-wide maximum across every shard - a selection matching only one // shard's rows must not claim exactness as of a sequence a different, untouched shard // happened to advance to. A selection matching no rows has no shard to scope to, so it // keeps the caller-supplied fallback. if (shardId != null && asOfForShard != null) { asOf = asOfForShard(shardId); } } var evaluation = CultNetSelectionEvaluator.Page(full, selection, asOf, cursorKey); var records = evaluation.Rows.Select(row => ToRawRecord(row, options)).ToArray(); var wantDocument = selection.Projection == CultNetSelectionProjections.Document; return new CultNetSelectionPage { // R-G/C5: the selection's whole matching count, not this page's row count. Matched = (uint)evaluation.TotalMatched, AsOf = asOf, Next = evaluation.NextCursor, Documents = wantDocument ? records : null, Headers = wantDocument ? null : records.Select(CultNetRawDocumentHeader.FromRecord).ToArray(), Edges = selection.HasHop ? evaluation.Edges.Select(edge => ToEdge(edge, wantDocument)).ToArray() : null }; } /// /// Refuses a selection whose matched rows come from more than one shard's mutation log (R-Q): /// asOf is one shard-log watermark ( is a single /// number today, not one per shard), so a page spanning two shards' logs cannot honestly claim /// to be exact as of it. Refused door-side, the same as an unreachable index or an empty /// keys list (R-F), rather than answered with an asOf that is not really defined - /// reusing /selection_invalid (R-N) rather /// than adding a second refusal shape for what is, from the wire's point of view, the same /// "this selection cannot be answered as asked" refusal a bad field or an empty list gets. /// /// Returns the one shard every matched row belongs to, or null when the selection matched no rows. private static string? EnsureSingleShard( IReadOnlyList rows, Func shardIdOf) { string? shardId = null; foreach (var row in rows) { var rowShardId = shardIdOf(row.Descriptor.SchemaId, row.Key); if (shardId == null) { shardId = rowShardId; } else if (!string.Equals(shardId, rowShardId, StringComparison.Ordinal)) { throw new CultNetSelectionInvalidException( "asOf", null, $"The selection matches rows from more than one shard's log (at least '{shardId}' and '{rowShardId}'); asOf is not defined for a shard-spanning page."); } } return shardId; } /// /// Evaluates a selection over every row the cache holds and projects every matched row - no /// cursor, no 200-row cap. The v0 lowering and the shard-bounded snapshot both answer their /// whole matching set in one message, so they call this once instead of paging through /// in a loop that re-filters and re-sorts the same row set per page /// (R-G: "answer v0 and shard paging from one evaluation"). /// internal CultNetSelectionPage SelectAll( CultCache cache, CultNetSelection selection, Func ordinalOf, ulong asOf, CultNetDocumentMessageOptions? options, Func? rowFilter = null) { selection = ExpandSchemaBindingAliases(selection); var stored = cache.AllStoredDocuments; if (rowFilter != null) stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key)); var rows = stored .Select(entry => new CultNetSelectionEvaluator.Row( entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt)) .ToArray(); var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection); var records = full.Ordered.Select(row => ToRawRecord(row, options)).ToArray(); var wantDocument = selection.Projection == CultNetSelectionProjections.Document; var edges = selection.HasHop ? CultNetSelectionEvaluator.EdgesFor(full.Ordered, full).Select(edge => ToEdge(edge, wantDocument)).ToArray() : null; return new CultNetSelectionPage { Matched = (uint)full.Ordered.Count, AsOf = asOf, Next = null, Documents = wantDocument ? records : null, Headers = wantDocument ? null : records.Select(CultNetRawDocumentHeader.FromRecord).ToArray(), Edges = edges }; } // A CultDocumentDescriptor knows nothing of CultNetDocumentBinding's optional schema-id // override; a caller filtering by that wire id (rather than the descriptor's own) needs the // descriptor's own id present in the selection too, since CultNetSelectionEvaluator only ever // reads descriptors. R-M: this is the one copy - both CultNetDatabaseServer's and // CultNetDatabaseSubscriptionServer's single-change fast path call this too instead of carrying // their own CultNetSchemaAliasMatching.WithBindingSchemaAlias (deleted). internal CultNetSelection ExpandSchemaBindingAliases(CultNetSelection selection) { if (selection.Schemas is not { Length: > 0 } || _bindingsByType.Count == 0) return selection; List? expanded = null; foreach (var binding in _bindingsByType.Values) { if (!selection.Schemas.Contains(binding.SchemaId, StringComparer.Ordinal)) continue; var descriptor = _documents.GetRequired(binding.DocumentType); if (selection.Schemas.Contains(descriptor.SchemaId, StringComparer.Ordinal)) continue; expanded ??= new List(selection.Schemas); if (!expanded.Contains(descriptor.SchemaId, StringComparer.Ordinal)) expanded.Add(descriptor.SchemaId); } if (expanded == null) return selection; return new CultNetSelection { Schemas = expanded.ToArray(), Keys = selection.Keys, Fields = selection.Fields, Cites = selection.Cites, Cited = selection.Cited, Projection = selection.Projection, Descending = selection.Descending, Limit = selection.Limit, Cursor = selection.Cursor }; } /// /// Builds the wire record for one evaluated row. The shared path under snapshot and change /// (docs/cultnet-selection-cut.md, section 8's "shared paths"). /// internal CultNetRawDocumentRecord ToRawRecord(CultNetSelectionEvaluator.Row row, CultNetDocumentMessageOptions? options = null) => ToRawRecord(row.Descriptor, row.Key, row.Document, string.IsNullOrEmpty(row.StoredAt) ? ResolveStoredAt(options) : row.StoredAt, options); internal CultNetRawDocumentRecord ToRawRecord( CultDocumentDescriptor descriptor, CultRecordKey key, object document, string storedAt, CultNetDocumentMessageOptions? options = null) { var binding = GetByDocumentType(descriptor.DocumentType) ?? new CultNetDocumentBinding( descriptor.DocumentType, descriptor.SchemaId, value => CultDocumentMessagePackSerialization.SerializeUntyped(value, value.GetType(), _documents), payload => CultDocumentMessagePackSerialization.DeserializeUntyped(descriptor.DocumentType, payload, _documents)); return new CultNetRawDocumentRecord { SchemaId = binding.SchemaId, SchemaName = descriptor.SchemaName, SchemaVersion = descriptor.SchemaVersion, SchemaContentHash = descriptor.ContentHash, RecordKey = key.Value, StoredAt = storedAt, PayloadEncoding = "messagepack", Payload = binding.PayloadSerializer(document), SourceRuntimeId = options?.SourceRuntimeId, SourceAgentId = options?.SourceAgentId, SourceRole = options?.SourceRole, Tags = options?.Tags }; } private CultNetEdge ToEdge(CultNetSelectionEvaluator.EdgeMatch edge, bool wantDocument) { // R-E: ToRawRecord and ToEdge emit the same id for the same row - the wire schema id a // binding overrides to (WireSchemaId), never the descriptor's own id when the two differ. var wire = new CultNetEdge { From = new CultNetRecordRef { SchemaId = WireSchemaId(edge.From.Descriptor), RecordKey = edge.From.Key.Value }, Role = edge.Role, To = new CultNetRecordRef { SchemaId = WireSchemaId(edge.To.Descriptor), RecordKey = edge.To.Key.Value } }; if (wantDocument && edge.Payload != null) { wire.PayloadEncoding = "messagepack"; // A dictionary reference's payload is typed by the member's declared value type (D11) - // routinely a plain value like float, not a registered CultDocument - so this serializes // it directly rather than through CultDocumentMessagePackSerialization.SerializeUntyped, // which requires the type to be a registered document and throws otherwise. wire.Payload = MessagePackSerializer.Serialize(edge.Payload.GetType(), edge.Payload, CultNetSchemaMessageSerialization.Options); } return wire; } /// /// The wire schema id this registry answers with for a descriptor - its binding's override when /// one is registered, else the descriptor's own id (R-E: the same id /// emits). /// Internal rather than private (R-P): one id reaches the authorizer, and it is this one, on /// every path - CultNetDatabaseSubscriptionServer's live path needs it too. /// internal string WireSchemaId(CultDocumentDescriptor descriptor) => GetByDocumentType(descriptor.DocumentType)?.SchemaId ?? descriptor.SchemaId; /// /// Applies a raw document put message to a cache. /// public async Task ApplyRawDocumentPutMessageAsync( CultCache cache, CultNetDocumentPutRawMessage message) { if (message.Document == null) { throw new ArgumentException("CultNet raw document message is missing its document payload.", nameof(message)); } var document = DeserializeRawDocument(message.Document); var descriptor = ResolveDescriptorForRawDocument(message.Document); var addMethod = typeof(CultCache).GetMethod(nameof(CultCache.AddAsync))! .MakeGenericMethod(descriptor.DocumentType); var handleType = typeof(CultRecordHandle<>).MakeGenericType(descriptor.DocumentType); var optionalHandle = Activator.CreateInstance(typeof(Nullable<>).MakeGenericType(handleType), new object[] { Activator.CreateInstance(handleType, new object[] { new CultRecordKey(message.Document.RecordKey) })! }); var task = (Task)addMethod.Invoke(cache, [document, optionalHandle])!; await task.ConfigureAwait(false); return document; } /// /// Decodes one validated raw document without materializing it in a CultCache replica. /// Use this for ephemeral reactive subscriptions whose callback is the consumer. /// public object DeserializeRawDocument(CultNetRawDocumentRecord document) { if (document == null) throw new ArgumentNullException(nameof(document)); ValidateRawDocumentRecord(document); var binding = GetBySchemaId(document.SchemaId); var descriptor = binding != null ? _documents.GetRequired(binding.DocumentType) : ResolveDescriptorForRawDocument(document); binding ??= new CultNetDocumentBinding( descriptor.DocumentType, descriptor.SchemaId, value => CultDocumentMessagePackSerialization.SerializeUntyped(value, value.GetType()), payload => CultDocumentMessagePackSerialization.DeserializeUntyped(descriptor.DocumentType, payload)); return binding.PayloadDeserializer(document.Payload); } /// /// Applies a raw document put message and returns the typed document. /// public async Task ApplyRawDocumentPutMessageAsync( CultCache cache, CultNetDocumentPutRawMessage message) where T : class { return (T)await ApplyRawDocumentPutMessageAsync(cache, message).ConfigureAwait(false); } /// /// Applies all documents from a raw snapshot response. /// public async Task> ApplyRawSnapshotResponseAsync( CultCache cache, CultNetSnapshotResponseRawMessage response) { var applied = new List(response.Documents.Length); foreach (var document in response.Documents) { applied.Add(await ApplyRawDocumentPutMessageAsync( cache, new CultNetDocumentPutRawMessage { MessageId = response.MessageId, Document = document }).ConfigureAwait(false)); } return applied; } internal CultDocumentDescriptor ResolveDescriptorForRawDocument(CultNetRawDocumentRecord document) { try { return ResolveDescriptorForSchemaId(document.SchemaId); } catch (InvalidOperationException) when (TryResolveDescriptorByPayloadSchema(document.Payload) is { } descriptor) { return descriptor; } } internal CultDocumentDescriptor ResolveDescriptorForSchemaId(string schemaId) { try { return _documents.GetRequiredBySchemaId(schemaId); } catch (InvalidOperationException) when (TryResolveDescriptorBySchemaAlias(schemaId) is { } descriptor) { return descriptor; } } private CultDocumentDescriptor? TryResolveDescriptorBySchemaAlias(string schemaId) { if (string.IsNullOrWhiteSpace(schemaId)) return null; var candidates = ResolvePayloadSchemaCandidates().ToArray(); var descriptor = candidates.FirstOrDefault(candidate => string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal) || string.Equals(candidate.SchemaName, schemaId, StringComparison.Ordinal) || string.Equals(candidate.SchemaVersion, schemaId, StringComparison.Ordinal) || candidate.ToCatalogEntry().CompatibleSchemaIds.Any(compatible => string.Equals(compatible, schemaId, StringComparison.Ordinal))); if (descriptor != null) return descriptor; var schemaName = InferSchemaName(schemaId); return string.IsNullOrWhiteSpace(schemaName) ? null : candidates.FirstOrDefault(candidate => string.Equals(candidate.SchemaName, schemaName, StringComparison.Ordinal)); } /// /// Resolves the descriptor this registry's own bindings (or every registered descriptor, when /// none are bound) associate with a raw payload's embedded schema version, when exactly one /// registered type owns that schema string. Used by /// to recover a document's type from a foreign/runtime-generated schema id. Ambiguous - and /// so not the right tool - when the caller already knows the target descriptor and two /// registered types alias the same schema id: match 's /// result against that descriptor with directly /// instead (docs/cultnet-selection-cut.md, section 4/S2-5; this is what CultMesh's /// foreign-schema decode does). Trusts the payload's own schema stamp; it does not re-run a /// selection or re-check authorization. /// public CultDocumentDescriptor? TryResolveDescriptorByPayloadSchema(byte[] payload) { var schemaVersion = TryReadSchemaVersion(payload); if (string.IsNullOrWhiteSpace(schemaVersion)) return null; var candidates = ResolvePayloadSchemaCandidates().ToArray(); var descriptor = candidates.FirstOrDefault(candidate => string.Equals(candidate.SchemaVersion, schemaVersion, StringComparison.Ordinal)); if (descriptor != null) return descriptor; var schemaName = InferSchemaName(schemaVersion!); return string.IsNullOrWhiteSpace(schemaName) ? null : candidates.FirstOrDefault(candidate => string.Equals(candidate.SchemaName, schemaName, StringComparison.Ordinal)); } private IEnumerable ResolvePayloadSchemaCandidates() { if (_bindingsBySchemaId.Count == 0 && _bindingsByType.Count == 0) return _documents.AllDescriptors; return _bindingsBySchemaId.Values .Concat(_bindingsByType.Values) .Select(binding => _documents.GetRequired(binding.DocumentType)) .GroupBy(descriptor => descriptor.DocumentType) .Select(group => group.First()); } /// /// True when a raw payload's embedded schemaVersion stamp (when it carries one) aliases /// the given descriptor - the "foreign schema id" fallback shared by CultMesh's peer-snapshot /// read and its typed-document decode (docs/cultnet-selection-cut.md, R-L: one registry method /// taking the descriptor, replacing two call sites that each read the payload and alias-matched /// it by hand). stays internal: a caller with an already-known /// descriptor never needs the raw schema-version string itself, only this answer. /// public static bool PayloadMatchesSchema(byte[] payload, CultDocumentDescriptor descriptor) => TryReadSchemaVersion(payload) is { } schemaVersion && CultNetSchemaAliasMatching.Matches(schemaVersion, descriptor); /// /// Reads a raw payload's embedded schemaVersion stamp, when it carries one - the parse /// step of the registry's payload-schema decode rule, shared with /// and /// (docs/cultnet-selection-cut.md, section 4/S2-5, R-L). Internal: a caller outside this /// assembly matching against one already-known descriptor uses ; /// resolving "whichever registered type owns this schema string" (ambiguous when two document /// types alias the same schema id) is 's job. /// internal static string? TryReadSchemaVersion(byte[] payload) { try { var array = MessagePackSerializer.Deserialize( payload, CultNetSchemaMessageSerialization.Options); if (array.Length > 0 && array[0] is string schemaVersion) return schemaVersion; } catch (Exception) { // Fall through to map decoding; different runtimes may encode object-like payloads. } try { var map = MessagePackSerializer.Deserialize>( payload, CultNetSchemaMessageSerialization.Options); if (map.TryGetValue("schemaVersion", out var schemaVersion) && schemaVersion is string schemaVersionText) return schemaVersionText; if (map.TryGetValue("schema_version", out var snakeSchemaVersion) && snakeSchemaVersion is string snakeSchemaVersionText) return snakeSchemaVersionText; } catch (Exception) { return null; } return null; } private static string? InferSchemaName(string schemaVersion) { var marker = schemaVersion.LastIndexOf(".v", StringComparison.Ordinal); if (marker <= 0 || marker + 2 >= schemaVersion.Length) return null; var version = schemaVersion.Substring(marker + 2); return version.All(char.IsDigit) ? schemaVersion.Substring(0, marker) : null; } /// /// Applies all typed documents from a raw snapshot response. /// public async Task> ApplyRawSnapshotResponseAsync( CultCache cache, CultNetSnapshotResponseRawMessage response) where T : class { return (await ApplyRawSnapshotResponseAsync(cache, response).ConfigureAwait(false)).OfType().ToArray(); } private static void ValidateRawDocumentRecord(CultNetRawDocumentRecord document) { if (document.PayloadEncoding != "messagepack") { throw new InvalidOperationException( $"CultNet raw document payloadEncoding must be \"messagepack\", not \"{document.PayloadEncoding}\"."); } RequireNonEmpty(document.SchemaId, nameof(document.SchemaId)); RequireNonEmpty(document.RecordKey, nameof(document.RecordKey)); RequireNonEmpty(document.StoredAt, nameof(document.StoredAt)); if (document.Payload == null || document.Payload.Length == 0) { throw new InvalidOperationException("CultNet raw document payload must be non-empty."); } } private static string ResolveStoredAt(CultNetDocumentMessageOptions? options) { return !string.IsNullOrWhiteSpace(options?.StoredAt) ? options!.StoredAt! : DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture); } private static string RequireNonEmpty(string value, string paramName) { if (string.IsNullOrWhiteSpace(value)) { throw new ArgumentException("Value must be non-empty.", paramName); } return value; } } }