From eaf1189cc13ec357cb54e2722de2793faee2680c Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Wed, 19 Aug 2026 13:22:36 +0200
Subject: [PATCH] Project subscription records per peer

---
 .../CultNetDatabaseSubscriptionServer.cs      | 20 ++++++++++++++++++-
 .../NetworkingTests.cs                        |  9 ++++++++-
 2 files changed, 27 insertions(+), 2 deletions(-)

diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
index cd224816..c1ca3c9d 100644
--- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
+++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs
@@ -13,6 +13,8 @@ namespace GameCult.Networking
 {
     /// <summary>
     /// Publishes live database changes through any schema-v0 server transport.
+    /// Optional record projection runs after authorization and owns both initial
+    /// snapshot and live-update delivery for a peer.
     /// </summary>
     public sealed class CultNetDatabaseSubscriptionServer : IDisposable
     {
@@ -22,6 +24,7 @@ namespace GameCult.Networking
         private readonly Func<CultNetDatabaseUnsubscribeMessage, ICultNetSchemaServerPeer, Task> _unsubscribe;
         private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, bool>? _authorizeRequest;
         private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? _authorizeRecord;
+        private readonly Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, CultNetRawDocumentRecord, CultNetRawDocumentRecord?>? _projectRecord;
         private readonly ICultNetSchemaServerPeerLifecycle? _peerLifecycle;
         private readonly ConcurrentDictionary<SubscriptionKey, IDisposable> _subscriptions =
             new ConcurrentDictionary<SubscriptionKey, IDisposable>(SubscriptionKeyComparer.Instance);
@@ -34,12 +37,14 @@ namespace GameCult.Networking
             ICultNetSchemaServer server,
             CultNetDatabase database,
             Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, bool>? authorizeRequest = null,
-            Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? authorizeRecord = null)
+            Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? authorizeRecord = null,
+            Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, CultNetRawDocumentRecord, CultNetRawDocumentRecord?>? projectRecord = null)
         {
             _server = server ?? throw new ArgumentNullException(nameof(server));
             _database = database ?? throw new ArgumentNullException(nameof(database));
             _authorizeRequest = authorizeRequest;
             _authorizeRecord = authorizeRecord;
+            _projectRecord = projectRecord;
             _subscribe = HandleSubscribeAsync;
             _unsubscribe = HandleUnsubscribeAsync;
             _server.OnCultNet(_subscribe);
@@ -134,6 +139,14 @@ namespace GameCult.Networking
                             request, peer, document.RecordKey, document.SchemaId))
                         .ToArray();
                 }
+                if (_projectRecord != null)
+                {
+                    snapshot.Documents = snapshot.Documents
+                        .Select(document => _projectRecord(request, peer, document))
+                        .Where(document => document != null)
+                        .Cast<CultNetRawDocumentRecord>()
+                        .ToArray();
+                }
                 peer.SendCultNet(snapshot);
             }
             else
@@ -205,6 +218,11 @@ namespace GameCult.Networking
             return _database.WatchAllChanges().Subscribe(change =>
             {
                 var message = CreateAuthorizedChange(change, request, subscriptionId, peer);
+                if (message?.Document != null && _projectRecord != null)
+                {
+                    message.Document = _projectRecord(request, peer, message.Document);
+                    if (message.Document == null) return;
+                }
                 if (message != null) peer.SendCultNet(message);
             });
         }
diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs
index ae3265ca..deac0d02 100644
--- a/tests/GameCult.Networking.Tests/NetworkingTests.cs
+++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs
@@ -1476,7 +1476,12 @@ namespace GameCult.Networking.Tests
                 server,
                 database,
                 authorizeRequest: (request, _) => request.ConsumerRuntimeId == "allowed-runtime",
-                authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal));
+                authorizeRecord: (_, _, recordKey, _) => recordKey.StartsWith("tests:public:", StringComparison.Ordinal),
+                projectRecord: (request, _, record) =>
+                {
+                    record.SourceRuntimeId = "projection:" + request.ConsumerRuntimeId;
+                    return record;
+                });
             using var cancellation = new CancellationTokenSource();
             var serverThread = new Thread(() =>
             {
@@ -1519,6 +1524,7 @@ namespace GameCult.Networking.Tests
             var snapshot = await AwaitWithTimeout(subscribed.Task, TimeSpan.FromSeconds(2));
             Assert.That(snapshot.Documents.Select(document => document.RecordKey),
                 Is.EqualTo(new[] { "tests:public:initial" }));
+            Assert.That(snapshot.Documents.Single().SourceRuntimeId, Is.EqualTo("projection:allowed-runtime"));
 
             await database.PutAsync(new CultRecordKey("tests:private:live"), new NetworkSchemaNote
             {
@@ -1535,6 +1541,7 @@ namespace GameCult.Networking.Tests
             cancellation.Cancel();
             Assert.That(update.Document, Is.Not.Null);
             Assert.That(update.Document!.RecordKey, Is.EqualTo("tests:public:live"));
+            Assert.That(update.Document.SourceRuntimeId, Is.EqualTo("projection:allowed-runtime"));
         }
 
         [Test]