From b54ed59135ad0ad7e97840824fafb7ee6a79ed52 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 17:10:24 +0100 Subject: [PATCH] Add CultNet database subscriptions --- .../Contracts/cultnet-distributed-database.md | 20 +-- src/GameCult.Networking/CultNetDatabase.cs | 9 ++ .../CultNetDatabaseServer.cs | 150 ++++++++++++++++++ .../CultNetSchemaMessageSerialization.cs | 3 + .../CultNetSchemaMessages.cs | 100 ++++++++++++ src/GameCult.Networking/README.md | 5 +- .../NetworkingTests.cs | 81 ++++++++++ 7 files changed, 356 insertions(+), 12 deletions(-) diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index 931f8bfc..ea4fbc2a 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -37,10 +37,11 @@ Implemented: - `CultNetDatabaseServer`: schema-v0 bridge for snapshot, put, and delete - `Client` and `Server` schema-v0 dispatch hooks - `CultNetHost.Database` and `CultNetHost.DatabaseServer` +- subscription request/cancel/change messages +- server-side live change fanout for schema/key-filtered subscriptions Not implemented yet: -- remote subscription request/stream messages - shard catalog exchange - membership/failure detection - leader election or automatic shard failover @@ -61,14 +62,13 @@ Not implemented yet: ## Next Coherent Slice -Add explicit subscription messages: +Add shard catalog exchange: -- subscribe by schema -- subscribe by record key -- subscribe by name/index -- unsubscribe -- stream domain changes as raw document put/delete records +- advertise shard ids, schema filters, key ranges/prefixes, owner runtime ids, + and epochs +- let clients ask any node where a write belongs +- reject stale-epoch writes explicitly +- prepare forwarding without making forwarding mandatory -That gives clients the RethinkDB/Firebase feel without requiring consensus or -membership work first. Shard catalogs and membership should follow once there -is more than one authority to discover. +That gives the subscription surface a real routing model before membership, +leader election, or replicated logs enter the room and start charging rent. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 0e36467d..bc1eeed6 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -390,6 +390,15 @@ namespace GameCult.Networking .Select(change => (CultNetDatabaseChange)change); } + /// + /// Watches all typed domain changes as boxed change objects. + /// + public Observable WatchAllChanges() + { + ThrowIfDisposed(); + return _changes; + } + /// /// Watches one record. /// diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index c7bc8a47..b46613c2 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -1,6 +1,11 @@ using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Reflection; using System.Threading.Tasks; +using GameCult.Caching; using LiteNetLib; +using R3; namespace GameCult.Networking { @@ -14,6 +19,9 @@ namespace GameCult.Networking private readonly Func _snapshotHandler; private readonly Func _putHandler; private readonly Func _deleteHandler; + private readonly Func _subscribeHandler; + private readonly Func _unsubscribeHandler; + private readonly ConcurrentDictionary _subscriptions = new(StringComparer.Ordinal); private bool _disposed; /// @@ -26,10 +34,14 @@ namespace GameCult.Networking _snapshotHandler = HandleSnapshotRequestAsync; _putHandler = HandlePutAsync; _deleteHandler = HandleDeleteAsync; + _subscribeHandler = HandleSubscribeAsync; + _unsubscribeHandler = HandleUnsubscribeAsync; _server.OnCultNet(_snapshotHandler); _server.OnCultNet(_putHandler); _server.OnCultNet(_deleteHandler); + _server.OnCultNet(_subscribeHandler); + _server.OnCultNet(_unsubscribeHandler); } /// @@ -79,6 +91,14 @@ namespace GameCult.Networking _server.RemoveCultNetMessageListener(_snapshotHandler); _server.RemoveCultNetMessageListener(_putHandler); _server.RemoveCultNetMessageListener(_deleteHandler); + _server.RemoveCultNetMessageListener(_subscribeHandler); + _server.RemoveCultNetMessageListener(_unsubscribeHandler); + foreach (var subscription in _subscriptions.Values) + { + subscription.Dispose(); + } + + _subscriptions.Clear(); } private Task HandleSnapshotRequestAsync(CultNetSnapshotRequestMessage request, NetPeer peer) @@ -110,5 +130,135 @@ namespace GameCult.Networking peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); } } + + private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage message, NetPeer peer) + { + var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId) + ? message.MessageId + : message.SubscriptionId; + if (string.IsNullOrWhiteSpace(subscriptionId)) + { + peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." }); + return Task.CompletedTask; + } + + var key = SubscriptionKey(peer, subscriptionId); + _subscriptions.AddOrUpdate( + key, + _ => CreateSubscription(message, subscriptionId, peer), + (_, existing) => + { + existing.Dispose(); + return CreateSubscription(message, subscriptionId, peer); + }); + + if (message.IncludeSnapshot) + { + peer.SendCultNet(CreateSnapshotResponse(new CultNetSnapshotRequestMessage + { + MessageId = message.MessageId, + SchemaIds = message.SchemaIds, + RecordKeys = message.RecordKeys + })); + } + + return Task.CompletedTask; + } + + private Task HandleUnsubscribeAsync(CultNetDatabaseUnsubscribeMessage message, NetPeer peer) + { + var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId) + ? message.MessageId + : message.SubscriptionId; + if (!string.IsNullOrWhiteSpace(subscriptionId) && + _subscriptions.TryRemove(SubscriptionKey(peer, subscriptionId), out var subscription)) + { + subscription.Dispose(); + } + + return Task.CompletedTask; + } + + private IDisposable CreateSubscription( + CultNetDatabaseSubscribeMessage request, + string subscriptionId, + NetPeer peer) + { + return _database.WatchAllChanges().Subscribe(change => + { + var outbound = CreateChangeMessage(change, subscriptionId, request); + if (outbound != null) + { + peer.SendCultNet(outbound); + } + }); + } + + internal CultNetDatabaseChangeRawMessage? CreateChangeMessage( + object change, + string subscriptionId, + CultNetDatabaseSubscribeMessage request) + { + var changeType = change.GetType(); + var key = (CultRecordKey)(changeType.GetProperty("Key")?.GetValue(change) ?? new CultRecordKey(string.Empty)); + var schemaId = (string?)changeType.GetProperty("SchemaId")?.GetValue(change) ?? string.Empty; + if (!Matches(request, schemaId, key)) + { + return null; + } + + var kind = (CultNetDatabaseChangeKind)(changeType.GetProperty("Kind")?.GetValue(change) ?? CultNetDatabaseChangeKind.Updated); + var document = changeType.GetProperty("Document")?.GetValue(change); + if (kind == CultNetDatabaseChangeKind.Removed || document == null) + { + return new CultNetDatabaseChangeRawMessage + { + MessageId = Guid.NewGuid().ToString("N"), + SubscriptionId = subscriptionId, + ChangeKind = "removed", + RecordKey = key.Value, + SchemaId = schemaId + }; + } + + return new CultNetDatabaseChangeRawMessage + { + MessageId = Guid.NewGuid().ToString("N"), + SubscriptionId = subscriptionId, + ChangeKind = kind == CultNetDatabaseChangeKind.Added ? "added" : "updated", + Document = CreateRawDocumentRecord(key, document) + }; + } + + private CultNetRawDocumentRecord CreateRawDocumentRecord(CultRecordKey key, object document) + { + var method = typeof(CultNetDocumentRegistry) + .GetMethods(BindingFlags.Public | BindingFlags.Instance) + .Single(candidate => candidate.Name == nameof(CultNetDocumentRegistry.CreateRawDocumentPutMessage) && + candidate.IsGenericMethodDefinition); + var documentType = document.GetType(); + var handleType = typeof(CultRecordHandle<>).MakeGenericType(documentType); + var handle = Activator.CreateInstance(handleType, new object[] { key }); + var message = method + .MakeGenericMethod(documentType) + .Invoke(_database.Documents, new[] { Guid.NewGuid().ToString("N"), handle, document, null }); + return ((CultNetDocumentPutRawMessage)message!).Document; + } + + private static bool Matches(CultNetDatabaseSubscribeMessage request, string schemaId, CultRecordKey key) + { + var schemaMatches = request.SchemaIds == null || + request.SchemaIds.Length == 0 || + request.SchemaIds.Contains(schemaId, StringComparer.Ordinal); + var keyMatches = request.RecordKeys == null || + request.RecordKeys.Length == 0 || + request.RecordKeys.Contains(key.Value, StringComparer.Ordinal); + return schemaMatches && keyMatches; + } + + private static string SubscriptionKey(NetPeer peer, string subscriptionId) + { + return $"{peer.Id}:{subscriptionId}"; + } } } diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index e48ed310..6abfe32b 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -59,6 +59,9 @@ namespace GameCult.Networking CultNetSchemaVersions.DocumentPutRaw => Deserialize(payload), CultNetSchemaVersions.SnapshotRequest => Deserialize(payload), CultNetSchemaVersions.SnapshotResponseRaw => Deserialize(payload), + CultNetSchemaVersions.DatabaseSubscribe => Deserialize(payload), + CultNetSchemaVersions.DatabaseUnsubscribe => Deserialize(payload), + CultNetSchemaVersions.DatabaseChangeRaw => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogRequest => Deserialize(payload), CultNetSchemaVersions.SchemaCatalogResponse => Deserialize(payload), _ => throw new MessagePackSerializationException( diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 5281fb79..21ca81db 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -79,6 +79,18 @@ namespace GameCult.Networking /// schema catalog response contract identifier. /// public const string SchemaCatalogResponse = "cultnet.schema_catalog_response.v0"; + /// + /// database subscription request contract identifier. + /// + public const string DatabaseSubscribe = "cultnet.database_subscribe.v0"; + /// + /// database subscription cancel contract identifier. + /// + public const string DatabaseUnsubscribe = "cultnet.database_unsubscribe.v0"; + /// + /// database subscription change contract identifier. + /// + public const string DatabaseChangeRaw = "cultnet.database_change_raw.v0"; } /// @@ -475,6 +487,94 @@ namespace GameCult.Networking [Key("documents")] public CultNetRawDocumentRecord[] Documents { get; set; } = Array.Empty(); } + /// + /// Requests a live database subscription. + /// + [MessagePackObject] + public class CultNetDatabaseSubscribeMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.DatabaseSubscribe; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets the live subscription id. + /// + [Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty; + /// + /// Gets or sets optional schema ids to watch. + /// + [Key("schemaIds")] public string[]? SchemaIds { get; set; } + /// + /// Gets or sets optional record keys to watch. + /// + [Key("recordKeys")] public string[]? RecordKeys { get; set; } + /// + /// Gets or sets whether the server should send a matching snapshot before live changes. + /// + [Key("includeSnapshot")] public bool IncludeSnapshot { get; set; } = true; + } + + /// + /// Cancels a live database subscription. + /// + [MessagePackObject] + public class CultNetDatabaseUnsubscribeMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.DatabaseUnsubscribe; + /// + /// Gets or sets the request id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets the live subscription id. + /// + [Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty; + } + + /// + /// Streams one raw document change for a live database subscription. + /// + [MessagePackObject] + public class CultNetDatabaseChangeRawMessage : ICultNetSchemaMessage + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.DatabaseChangeRaw; + /// + /// Gets or sets the message id. + /// + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + /// + /// Gets or sets the live subscription id. + /// + [Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty; + /// + /// Gets or sets the change kind. + /// + [Key("changeKind")] public string ChangeKind { get; set; } = string.Empty; + /// + /// Gets or sets the current document for added or updated changes. + /// + [Key("document")] public CultNetRawDocumentRecord? Document { get; set; } + /// + /// Gets or sets the deleted record key for removed changes. + /// + [Key("recordKey")] public string? RecordKey { get; set; } + /// + /// Gets or sets the deleted schema id for removed changes. + /// + [Key("schemaId")] public string? SchemaId { get; set; } + } + /// /// Describes one schema exposed through the CultNet schema catalog. /// diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 1165b1b9..4b8c6870 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -131,8 +131,9 @@ CultNet becomes the distribution model: The current first slice exposes this as `CultNetDatabase` and wires hosted servers through `CultNetDatabaseServer`. It handles primary-shard writes, typed -R3 watch streams, raw document puts/deletes, and raw snapshot responses. Remote -subscription messages and shard catalog exchange are the next missing organs. +R3 watch streams, raw document puts/deletes, raw snapshot responses, and +schema/key-filtered live subscriptions. Shard catalog exchange is the next +missing organ. The first coherent shard policy is primary ownership. Each shard has one authoritative writer at a time. Clients may connect to any node; a non-owner diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 1fe5247e..971bfd51 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -214,6 +214,33 @@ namespace GameCult.Networking.Tests Assert.That(roundTrip.Documents[0].Tags, Is.EqualTo(["swarm", "dream"])); } + [Test] + public void CultNetSchemaMessageSerialization_RoundTrips_DatabaseChangeRaw() + { + var message = new CultNetDatabaseChangeRawMessage + { + MessageId = "change-1", + SubscriptionId = "sub-1", + ChangeKind = "updated", + Document = new CultNetRawDocumentRecord + { + SchemaId = "schema-1", + RecordKey = "record-1", + StoredAt = "2026-05-19T12:00:00.0000000+00:00", + PayloadEncoding = "messagepack", + Payload = [0x91, 0x01] + } + }; + + var payload = CultNetSchemaMessageSerialization.Serialize(message); + var roundTrip = (CultNetDatabaseChangeRawMessage)CultNetSchemaMessageSerialization.Deserialize(payload); + + Assert.That(roundTrip.SubscriptionId, Is.EqualTo("sub-1")); + Assert.That(roundTrip.ChangeKind, Is.EqualTo("updated")); + Assert.That(roundTrip.Document, Is.Not.Null); + Assert.That(roundTrip.Document!.RecordKey, Is.EqualTo("record-1")); + } + [Test] public async Task CultNetDocumentRegistry_RawSnapshotReplication_PreservesPayloadBytes() { @@ -389,6 +416,60 @@ namespace GameCult.Networking.Tests Assert.That(response.Documents[0].Payload, Is.EqualTo(SerializePlayerDataPayload(player))); } + [Test] + public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() + { + var cache = new CultCache(); + var registry = new CultNetDocumentRegistry(cache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + cache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry + }); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, database); + var key = new CultRecordKey("player:change"); + var player = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "change@example.test", + PasswordHash = "hash", + Username = "Change" + }; + var schemaId = cache.Registry.GetRequired().SchemaId; + var change = new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Added, + key, + schemaId, + database.Shards[0], + player, + previousDocument: null); + var method = typeof(CultNetDatabaseServer).GetMethod( + "CreateChangeMessage", + BindingFlags.Instance | BindingFlags.NonPublic)!; + + var message = (CultNetDatabaseChangeRawMessage?)method.Invoke(databaseServer, new object[] + { + change, + "sub-1", + new CultNetDatabaseSubscribeMessage + { + SubscriptionId = "sub-1", + SchemaIds = [schemaId], + RecordKeys = [key.Value] + } + }); + + Assert.That(message, Is.Not.Null); + Assert.That(message!.SubscriptionId, Is.EqualTo("sub-1")); + Assert.That(message.ChangeKind, Is.EqualTo("added")); + Assert.That(message.Document, Is.Not.Null); + Assert.That(message.Document!.Payload, Is.EqualTo(SerializePlayerDataPayload(player))); + } + [Test] public void CultNetSchemaRegistry_BuiltInCatalog_AdvertisesRawLane_AndSharedGhostlightContract() {