diff --git a/src/GameCult.Networking/Client.cs b/src/GameCult.Networking/Client.cs index c58357f8..67ca9fe5 100644 --- a/src/GameCult.Networking/Client.cs +++ b/src/GameCult.Networking/Client.cs @@ -52,6 +52,7 @@ namespace GameCult.Networking private int _reconnectAttemptCount; private readonly ConcurrentDictionary _messageDelegates = new(); + private readonly ConcurrentDictionary _cultNetMessageDelegates = new(); private string? _sessionToken; private string _lastHost = "localhost"; @@ -137,12 +138,43 @@ namespace GameCult.Networking else Logger.LogError("Cannot send, client is not connected!"); } + /// + /// Sends a modern CultNet schema-v0 message when the client is connected and authorized. + /// + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (_disposed) + { + throw new ObjectDisposedException(nameof(Client)); + } + + if (message == null) throw new ArgumentNullException(nameof(message)); + + if (Connected && _peer != null) + { + if (Verified || message is CultNetLoginMessage or CultNetRegisterMessage or CultNetVerifyMessage) + { + Logger.LogDebug($"Sending CultNet schema message {message.SchemaVersion}"); + _peer.SendCultNet(message); + } + else + { + Logger.LogError("Cannot send CultNet schema message, client is not verified!"); + } + } + else + { + Logger.LogError("Cannot send CultNet schema message, client is not connected!"); + } + } + /// /// Removes all registered message listeners. /// public void ClearMessageListeners() { _messageDelegates.Clear(); + _cultNetMessageDelegates.Clear(); } /// @@ -172,6 +204,29 @@ namespace GameCult.Networking AddMessageListener(callback); } + /// + /// Adds a listener for a modern CultNet schema-v0 message type. + /// + public void AddCultNetMessageListener(Action callback) where T : ICultNetSchemaMessage + { + var type = typeof(T); + _cultNetMessageDelegates.AddOrUpdate(type, + _ => callback, + (t, current) => + { + var combined = Delegate.Combine(current, callback) as Action; + return combined ?? throw new InvalidOperationException($"Failed to combine delegates for {t.Name}"); + }); + } + + /// + /// Adds a listener for a modern CultNet schema-v0 message type. + /// + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + AddCultNetMessageListener(callback); + } + /// /// Removes a previously registered listener for a specific message type. /// @@ -194,6 +249,26 @@ namespace GameCult.Networking RemoveMessageListener(callback); } + /// + /// Removes a previously registered modern CultNet schema-v0 listener. + /// + public void RemoveCultNetMessageListener(Action callback) where T : ICultNetSchemaMessage + { + if (_cultNetMessageDelegates.TryGetValue(typeof(T), out var currentDelegate)) + { + var newDelegate = Delegate.Remove(currentDelegate, callback) as Action; + _cultNetMessageDelegates[typeof(T)] = newDelegate ?? (_ => { }); + } + } + + /// + /// Removes a previously registered modern CultNet schema-v0 listener. + /// + public void OffCultNet(Action callback) where T : ICultNetSchemaMessage + { + RemoveCultNetMessageListener(callback); + } + /// /// Sends a login request. /// @@ -293,6 +368,23 @@ namespace GameCult.Networking try { var bytes = reader.GetRemainingBytes(); + if (TryDeserializeCultNet(bytes, out var cultNetMessage)) + { + Logger.LogDebug($"Received CultNet schema message {cultNetMessage.SchemaVersion}"); + var cultNetType = cultNetMessage.GetType(); + if (_cultNetMessageDelegates.TryGetValue(cultNetType, out var cultNetDelegate) && + cultNetDelegate != null) + { + cultNetDelegate.DynamicInvoke(cultNetMessage); + } + else + { + Logger.LogWarning($"No listener for CultNet schema message {cultNetMessage.SchemaVersion}"); + } + + return; + } + var message = MessageSerialization.Deserialize(bytes); if (LogSensitivePayloads) { @@ -355,6 +447,20 @@ namespace GameCult.Networking (peer, latency) => Ping = latency; //Logger($"Ping received: {latency} ms"); } + private static bool TryDeserializeCultNet(byte[] payload, out ICultNetSchemaMessage message) + { + try + { + message = CultNetSchemaMessageSerialization.Deserialize(payload); + return true; + } + catch (MessagePackSerializationException) + { + message = null!; + return false; + } + } + /// /// Disconnects the client and suppresses automatic reconnect. /// diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md new file mode 100644 index 00000000..931f8bfc --- /dev/null +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -0,0 +1,74 @@ +# CultNet Distributed Database Notes + +CultNet's distributed database path is a typed realtime database over CultCache +documents. The design should borrow proven distributed-systems ideas without +copying their entire machine or laundering vague consistency through impressive +words. + +## Prior Art To Keep + +- RethinkDB changefeeds: subscriptions should emit old/new document views and + support point, table/schema, and filtered watches. +- Firebase Realtime Database: client ergonomics should make realtime sync feel + like the default path, but presence, disconnect handling, and server-side + authority still need explicit ownership. +- Dynamo: partitioning, replication metadata, vector-ish causality, hinted + recovery, and application-visible conflict resolution are useful. "Always + writeable" is not the first CultNet promise. +- Raft: primary authority per shard is the first understandable consistency + model. Leader election/log replication can come later if the cluster needs + automatic failover. +- SWIM: peer membership and failure suspicion should be gossip-shaped when the + mesh grows beyond small static clusters. +- CRDTs: offline multi-writer documents are valid only for types with declared, + deterministic merge semantics. +- Hashgraph: gossip-about-gossip is interesting for compact event provenance + and fair ordering research. CultNet should not adopt public-ledger machinery + or crypto-token assumptions as a shortcut to database correctness. + +## Current Cut + +Implemented: + +- `CultNetDatabase`: database-style facade over a `CultCache` +- primary-shard write checks +- typed R3 watch streams +- raw CultNet put/delete application through shard policy +- `CultNetDatabaseServer`: schema-v0 bridge for snapshot, put, and delete +- `Client` and `Server` schema-v0 dispatch hooks +- `CultNetHost.Database` and `CultNetHost.DatabaseServer` + +Not implemented yet: + +- remote subscription request/stream messages +- shard catalog exchange +- membership/failure detection +- leader election or automatic shard failover +- replicated log +- declared CRDT merge policies + +## Live Invariants + +- CultCache owns document schema, identity, local indexes, and local diffing. +- CultNet owns transport, shard authority, remote mutation delivery, and + subscription fanout. +- One primary owns writes for a shard until a stronger policy exists. +- Non-owner writes are rejected or explicitly forwarded. They are not silently + applied. +- Raw snapshot/put/delete messages must pass through `CultNetDatabase`, not + bypass shard policy. +- Realtime subscriptions publish domain changes, not storage envelopes. + +## Next Coherent Slice + +Add explicit subscription messages: + +- subscribe by schema +- subscribe by record key +- subscribe by name/index +- unsubscribe +- stream domain changes as raw document put/delete records + +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. diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs new file mode 100644 index 00000000..c7bc8a47 --- /dev/null +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -0,0 +1,114 @@ +using System; +using System.Threading.Tasks; +using LiteNetLib; + +namespace GameCult.Networking +{ + /// + /// Routes CultNet schema-v0 document messages through a . + /// + public sealed class CultNetDatabaseServer : IDisposable + { + private readonly Server _server; + private readonly CultNetDatabase _database; + private readonly Func _snapshotHandler; + private readonly Func _putHandler; + private readonly Func _deleteHandler; + private bool _disposed; + + /// + /// Creates and attaches a database message bridge to a server. + /// + public CultNetDatabaseServer(Server server, CultNetDatabase database) + { + _server = server ?? throw new ArgumentNullException(nameof(server)); + _database = database ?? throw new ArgumentNullException(nameof(database)); + _snapshotHandler = HandleSnapshotRequestAsync; + _putHandler = HandlePutAsync; + _deleteHandler = HandleDeleteAsync; + + _server.OnCultNet(_snapshotHandler); + _server.OnCultNet(_putHandler); + _server.OnCultNet(_deleteHandler); + } + + /// + /// Gets the database used by this bridge. + /// + public CultNetDatabase Database => _database; + + /// + /// Applies a raw put message through the database shard policy. + /// + public Task ApplyPutAsync(CultNetDocumentPutRawMessage message) + { + return _database.ApplyPutAsync(message); + } + + /// + /// Applies a raw delete message through the database shard policy. + /// + public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) + { + return _database.ApplyDeleteAsync(message); + } + + /// + /// Creates a raw snapshot response from the database cache. + /// + public CultNetSnapshotResponseRawMessage CreateSnapshotResponse(CultNetSnapshotRequestMessage request) + { + if (request == null) throw new ArgumentNullException(nameof(request)); + return _database.Documents.CreateRawSnapshotResponse( + _database.Cache, + string.IsNullOrWhiteSpace(request.MessageId) ? Guid.NewGuid().ToString("N") : request.MessageId, + request); + } + + /// + /// Detaches handlers from the server. + /// + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _server.RemoveCultNetMessageListener(_snapshotHandler); + _server.RemoveCultNetMessageListener(_putHandler); + _server.RemoveCultNetMessageListener(_deleteHandler); + } + + private Task HandleSnapshotRequestAsync(CultNetSnapshotRequestMessage request, NetPeer peer) + { + peer.SendCultNet(CreateSnapshotResponse(request)); + return Task.CompletedTask; + } + + private async Task HandlePutAsync(CultNetDocumentPutRawMessage message, NetPeer peer) + { + try + { + await ApplyPutAsync(message).ConfigureAwait(false); + } + catch (Exception ex) + { + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); + } + } + + private async Task HandleDeleteAsync(CultNetDocumentDeleteMessage message, NetPeer peer) + { + try + { + await ApplyDeleteAsync(message).ConfigureAwait(false); + } + catch (Exception ex) + { + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); + } + } + } +} diff --git a/src/GameCult.Networking/CultNetLocal.cs b/src/GameCult.Networking/CultNetLocal.cs index 9fe5f1fd..1227aa6f 100644 --- a/src/GameCult.Networking/CultNetLocal.cs +++ b/src/GameCult.Networking/CultNetLocal.cs @@ -42,12 +42,18 @@ namespace GameCult.Networking /// public sealed class CultNetHost : IDisposable { - internal CultNetHost(CultCache cache, SingleFileMessagePackBackingStore? store, Server server, CultNetDatabase database) + internal CultNetHost( + CultCache cache, + SingleFileMessagePackBackingStore? store, + Server server, + CultNetDatabase database, + CultNetDatabaseServer databaseServer) { Cache = cache; Store = store; Server = server; Database = database; + DatabaseServer = databaseServer; } /// @@ -70,6 +76,11 @@ namespace GameCult.Networking /// public CultNetDatabase Database { get; } + /// + /// Gets the schema-v0 server bridge for distributed database messages. + /// + public CultNetDatabaseServer DatabaseServer { get; } + /// /// Starts the server. /// @@ -91,6 +102,7 @@ namespace GameCult.Networking /// public void Dispose() { + DatabaseServer.Dispose(); Server.Dispose(); Database.Dispose(); Cache.Dispose(); @@ -118,9 +130,10 @@ namespace GameCult.Networking var server = new Server(cache, options.Security ?? ServerSecurityOptions.Development()); var databaseOptions = options.DatabaseOptions ?? new CultNetDatabaseOptions(); var database = new CultNetDatabase(cache, databaseOptions); + var databaseServer = new CultNetDatabaseServer(server, database); options.ConfigureServer?.Invoke(server); - var host = new CultNetHost(cache, store, server, database); + var host = new CultNetHost(cache, store, server, database, databaseServer); if (options.StartServer) { host.Start(); diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index 484128fb..1165b1b9 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -57,6 +57,8 @@ Modern schema-v0 helpers live in: - `CultNetSchemaMessageSerialization` - `CultNetSchemaRegistry` - `CultNetDocumentRegistry` +- `CultNetDatabase` +- `CultNetDatabaseServer` - `NetPeerExtensions.SendCultNet(...)` ## Main Types @@ -127,6 +129,11 @@ CultNet becomes the distribution model: - remote subscription fanout - reconnect resynchronization +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. + 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 node forwards a write to the owner or rejects it honestly when forwarding is diff --git a/src/GameCult.Networking/Server.cs b/src/GameCult.Networking/Server.cs index 1d703185..e098ea25 100644 --- a/src/GameCult.Networking/Server.cs +++ b/src/GameCult.Networking/Server.cs @@ -29,6 +29,7 @@ namespace GameCult.Networking private const int MaxLoginAttemptsPerMinute = 5; private readonly ConcurrentDictionary _messageDelegates = new(); + private readonly ConcurrentDictionary _cultNetMessageDelegates = new(); private readonly ConcurrentDictionary _users = new(); private readonly ConcurrentDictionary> _connectionAttempts = new(); private readonly ConcurrentDictionary _connectionAttemptLocks = new(); @@ -92,6 +93,7 @@ namespace GameCult.Networking public void ClearMessageListeners() { _messageDelegates.Clear(); + _cultNetMessageDelegates.Clear(); } /// @@ -119,6 +121,49 @@ namespace GameCult.Networking AddMessageListener(callback); } + /// + /// Adds a listener for a modern CultNet schema-v0 message type. + /// + public void AddCultNetMessageListener(Func callback) where T : ICultNetSchemaMessage + { + var type = typeof(T); + _cultNetMessageDelegates.AddOrUpdate(type, + _ => callback, + (t, current) => + { + var combined = Delegate.Combine(current, callback) as Func; + return combined ?? throw new InvalidOperationException($"Failed to combine delegates for {t.Name}"); + }); + } + + /// + /// Adds a listener for a modern CultNet schema-v0 message type. + /// + public void AddCultNetMessageListener(Action callback) where T : ICultNetSchemaMessage + { + AddCultNetMessageListener((message, peer) => + { + callback(message, peer); + return Task.CompletedTask; + }); + } + + /// + /// Adds a listener for a modern CultNet schema-v0 message type. + /// + public void OnCultNet(Func callback) where T : ICultNetSchemaMessage + { + AddCultNetMessageListener(callback); + } + + /// + /// Adds a listener for a modern CultNet schema-v0 message type. + /// + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + AddCultNetMessageListener(callback); + } + /// /// Removes a previously registered listener for a specific message type. /// @@ -141,6 +186,25 @@ namespace GameCult.Networking RemoveMessageListener(callback); } + /// + /// Removes a previously registered modern CultNet schema-v0 listener. + /// + public void RemoveCultNetMessageListener(Delegate callback) where T : ICultNetSchemaMessage + { + if (_cultNetMessageDelegates.TryGetValue(typeof(T), out var currentDelegate)) + { + var newDelegate = Delegate.Remove(currentDelegate, callback); + if (newDelegate == null) + { + _cultNetMessageDelegates.TryRemove(typeof(T), out _); + } + else + { + _cultNetMessageDelegates[typeof(T)] = newDelegate; + } + } + } + /// /// Stops the underlying LiteNetLib server. /// @@ -210,6 +274,14 @@ namespace GameCult.Networking try { var bytes = reader.GetRemainingBytes(); + var user = _users.GetOrAdd(peer.Id, _ => new User { Peer = peer }); + if (TryDeserializeCultNet(bytes, out var cultNetMessage)) + { + Logger.LogDebug($"Received CultNet schema message {cultNetMessage.SchemaVersion}"); + await HandleCultNetSchemaMessageAsync(peer, user, cultNetMessage).ConfigureAwait(false); + return; + } + var message = MessageSerialization.Deserialize(bytes); if (LogSensitivePayloads) { @@ -225,8 +297,6 @@ namespace GameCult.Networking } message.Peer = peer; - var user = _users.GetOrAdd(peer.Id, _ => new User { Peer = peer }); - if (message is LoginMessage or RegisterMessage or VerifyMessage) { if (IsVerified(user)) @@ -310,6 +380,53 @@ namespace GameCult.Networking Logger.LogInfo($"Server started on port {ServerPort}."); } + private async Task HandleCultNetSchemaMessageAsync(NetPeer peer, User user, ICultNetSchemaMessage message) + { + if (!IsVerified(user) && !CanProcessBeforeVerification(message)) + { + peer.SendCultNet(new CultNetErrorMessage { Error = "User Not Verified" }); + return; + } + + if (_cultNetMessageDelegates.TryGetValue(message.GetType(), out var del) && del != null) + { + foreach (var listener in del.GetInvocationList()) + { + var result = listener.DynamicInvoke(message, peer); + if (result is Task task) + { + await task.ConfigureAwait(false); + } + } + + user.SessionExpiresAt = DateTimeOffset.UtcNow.AddSeconds(SessionTimeoutSeconds); + RefreshSessionIfNeeded(peer, user); + } + else + { + Logger.LogWarning($"No listener for CultNet schema message {message.SchemaVersion}"); + } + } + + private static bool CanProcessBeforeVerification(ICultNetSchemaMessage message) + { + return message is CultNetHelloMessage or CultNetSchemaCatalogRequestMessage; + } + + private static bool TryDeserializeCultNet(byte[] payload, out ICultNetSchemaMessage message) + { + try + { + message = CultNetSchemaMessageSerialization.Deserialize(payload); + return true; + } + catch (MessagePackSerializationException) + { + message = null!; + return false; + } + } + /// public void Dispose() { diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index de6e8352..1fe5247e 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -355,6 +355,40 @@ namespace GameCult.Networking.Tests Assert.That(changes[0].Document!.Username, Is.EqualTo("Raw")); } + [Test] + public async Task CultNetDatabaseServer_Creates_Filtered_SnapshotResponse() + { + 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 player = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "snapshot@example.test", + PasswordHash = "hash", + Username = "Snapshot" + }; + var handle = await database.PutAsync(new CultRecordKey("player:snapshot"), player); + + var response = databaseServer.CreateSnapshotResponse(registry.CreateSnapshotRequest( + "snapshot-request", + recordKeys: [handle.Key.Value])); + + Assert.That(response.MessageId, Is.EqualTo("snapshot-request")); + Assert.That(response.Documents, Has.Length.EqualTo(1)); + Assert.That(response.Documents[0].RecordKey, Is.EqualTo(handle.Key.Value)); + Assert.That(response.Documents[0].Payload, Is.EqualTo(SerializePlayerDataPayload(player))); + } + [Test] public void CultNetSchemaRegistry_BuiltInCatalog_AdvertisesRawLane_AndSharedGhostlightContract() {