diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
new file mode 100644
index 00000000..151722d6
--- /dev/null
+++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs
@@ -0,0 +1,190 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Threading.Tasks;
+using GameCult.Caching;
+
+namespace GameCult.Networking
+{
+ ///
+ /// Replicates selected remote database records into a local typed CultCache over one retained session.
+ ///
+ public sealed class CultNetDatabaseSubscriptionClient : IDisposable
+ {
+ private readonly ICultNetSchemaClient _client;
+ private readonly CultCache _cache;
+ private readonly CultNetDocumentRegistry _documents;
+ private readonly ConcurrentDictionary>> _initialSnapshots = new();
+ private readonly object _queueLock = new();
+ private Task _queue = Task.CompletedTask;
+ private bool _disposed;
+
+ public CultNetDatabaseSubscriptionClient(
+ ICultNetSchemaClient client,
+ CultCache cache,
+ CultNetDocumentRegistry documents)
+ {
+ _client = client ?? throw new ArgumentNullException(nameof(client));
+ _cache = cache ?? throw new ArgumentNullException(nameof(cache));
+ _documents = documents ?? throw new ArgumentNullException(nameof(documents));
+ _client.OnCultNet(HandleSnapshot);
+ _client.OnCultNet(HandleChange);
+ }
+
+ /// Raised after a live change has been applied to the local typed cache.
+ public event Action? Changed;
+
+ /// Gets the local reactive cache receiving replicated documents.
+ public CultCache Cache => _cache;
+
+ ///
+ /// Starts a remote subscription and waits until its initial matching snapshot is in the local cache.
+ ///
+ public Task> SubscribeAsync(
+ string subscriptionId,
+ IEnumerable? recordKeys = null,
+ IEnumerable? schemaIds = null)
+ {
+ if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient));
+ if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing.");
+ if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId));
+
+ var messageId = Guid.NewGuid().ToString("N");
+ var completion = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously);
+ if (!_initialSnapshots.TryAdd(messageId, completion))
+ throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'.");
+
+ _client.SendCultNet(new CultNetDatabaseSubscribeMessage
+ {
+ MessageId = messageId,
+ SubscriptionId = subscriptionId,
+ RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
+ SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
+ IncludeSnapshot = true
+ });
+ return completion.Task;
+ }
+
+ /// Stops one live remote subscription.
+ public void Unsubscribe(string subscriptionId)
+ {
+ if (_disposed) return;
+ if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId));
+ _client.SendCultNet(new CultNetDatabaseUnsubscribeMessage
+ {
+ MessageId = Guid.NewGuid().ToString("N"),
+ SubscriptionId = subscriptionId
+ });
+ }
+
+ public void Dispose()
+ {
+ if (_disposed) return;
+ _disposed = true;
+ foreach (var completion in _initialSnapshots.Values)
+ completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)));
+ _initialSnapshots.Clear();
+ _client.Dispose();
+ }
+
+ private void HandleSnapshot(CultNetSnapshotResponseRawMessage message)
+ {
+ Enqueue(async () =>
+ {
+ if (!_initialSnapshots.TryRemove(message.MessageId, out var completion)) return;
+ try
+ {
+ var applied = await _documents.ApplyRawSnapshotResponseAsync(_cache, message).ConfigureAwait(false);
+ completion.TrySetResult(applied);
+ }
+ catch (Exception error)
+ {
+ completion.TrySetException(error);
+ }
+ });
+ }
+
+ private void HandleChange(CultNetDatabaseChangeRawMessage message)
+ {
+ Enqueue(async () =>
+ {
+ if (string.Equals(message.ChangeKind, "removed", StringComparison.Ordinal))
+ {
+ await DeleteAsync(message).ConfigureAwait(false);
+ Changed?.Invoke(new CultNetReplicatedDocumentChange(
+ message.SubscriptionId,
+ "removed",
+ message.RecordKey ?? string.Empty,
+ message.SchemaId ?? string.Empty,
+ null));
+ return;
+ }
+
+ if (message.Document == null) return;
+ var document = await _documents.ApplyRawDocumentPutMessageAsync(
+ _cache,
+ new CultNetDocumentPutRawMessage
+ {
+ MessageId = message.MessageId,
+ Document = message.Document
+ }).ConfigureAwait(false);
+ Changed?.Invoke(new CultNetReplicatedDocumentChange(
+ message.SubscriptionId,
+ message.ChangeKind,
+ message.Document.RecordKey,
+ message.Document.SchemaId,
+ document));
+ });
+ }
+
+ private async Task DeleteAsync(CultNetDatabaseChangeRawMessage message)
+ {
+ if (string.IsNullOrWhiteSpace(message.RecordKey) || string.IsNullOrWhiteSpace(message.SchemaId)) return;
+ var descriptor = _documents.ResolveDescriptorForSchemaId(message.SchemaId!);
+ var handle = Activator.CreateInstance(
+ typeof(CultRecordHandle<>).MakeGenericType(descriptor.DocumentType),
+ new object[] { new CultRecordKey(message.RecordKey!) });
+ var task = (Task)typeof(CultCache)
+ .GetMethod(nameof(CultCache.DeleteAsync), BindingFlags.Public | BindingFlags.Instance)!
+ .MakeGenericMethod(descriptor.DocumentType)
+ .Invoke(_cache, new[] { handle })!;
+ await task.ConfigureAwait(false);
+ }
+
+ private void Enqueue(Func operation)
+ {
+ lock (_queueLock)
+ {
+ _queue = _queue.ContinueWith(
+ _ => operation(),
+ TaskScheduler.Default).Unwrap();
+ }
+ }
+ }
+
+ /// Describes one typed document change after local replication.
+ public sealed class CultNetReplicatedDocumentChange
+ {
+ public CultNetReplicatedDocumentChange(
+ string subscriptionId,
+ string changeKind,
+ string recordKey,
+ string schemaId,
+ object? document)
+ {
+ SubscriptionId = subscriptionId;
+ ChangeKind = changeKind;
+ RecordKey = recordKey;
+ SchemaId = schemaId;
+ Document = document;
+ }
+
+ public string SubscriptionId { get; }
+ public string ChangeKind { get; }
+ public string RecordKey { get; }
+ public string SchemaId { get; }
+ public object? Document { get; }
+ }
+}
diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs
index 82f777f5..42707754 100644
--- a/tests/GameCult.Networking.Tests/NetworkingTests.cs
+++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs
@@ -1362,6 +1362,63 @@ namespace GameCult.Networking.Tests
Assert.That(update.Document!.RecordKey, Is.EqualTo(recordKey));
}
+ [Test]
+ public async Task DatabaseSubscriptionClient_ReplicatesInitialAndLiveTypedDocumentsOverRudp()
+ {
+ var sourceCache = new CultCache();
+ var sourceDatabase = new CultNetDatabase(sourceCache);
+ const string recordKey = "tests:subscription-client:note";
+ await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
+ {
+ Schema = "tests.networking_note.v1",
+ Text = "initial"
+ });
+ using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+ {
+ RuntimeId = "database-subscription-client-server",
+ Socket = BindUdpSocket()
+ });
+ using var subscriptions = new CultNetDatabaseSubscriptionServer(server, sourceDatabase);
+ using var cancellation = new CancellationTokenSource();
+ var serverThread = new Thread(() =>
+ {
+ while (!cancellation.IsCancellationRequested)
+ {
+ _ = server.PollOnceAsync().GetAwaiter().GetResult();
+ Thread.Sleep(1);
+ }
+ }) { IsBackground = true };
+ serverThread.Start();
+
+ var targetCache = new CultCache();
+ var targetDocuments = new CultNetDocumentRegistry(targetCache.Registry)
+ .Register(CultNetDocumentBinding.ForDocument(targetCache.Registry));
+ var transport = CultNetSchemaClients.CreateRudp("database-subscription-client");
+ using var client = new CultNetDatabaseSubscriptionClient(transport, targetCache, targetDocuments);
+ transport.Connect("127.0.0.1", server.LocalEndPoint.Port);
+ await WaitUntilAsync(() => transport.Connected, TimeSpan.FromSeconds(2));
+
+ var initial = await AwaitWithTimeout(
+ client.SubscribeAsync("notes", recordKeys: new[] { recordKey }),
+ TimeSpan.FromSeconds(2));
+ Assert.That(initial.OfType().Single().Text, Is.EqualTo("initial"));
+ Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.TypeOf());
+
+ var changed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ client.Changed += change => changed.TrySetResult(change);
+ await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
+ {
+ Schema = "tests.networking_note.v1",
+ Text = "live"
+ });
+ var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2));
+
+ cancellation.Cancel();
+ Assert.That(update.SubscriptionId, Is.EqualTo("notes"));
+ Assert.That(update.Document, Is.TypeOf());
+ Assert.That(((NetworkSchemaNote)targetCache.Get(new CultRecordKey(recordKey))!).Text, Is.EqualTo("live"));
+ }
+
[Test]
public async Task RudpCultNetSchemaServer_DeliversLargeSnapshotResponse()
{