GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Resolve database raw puts by payload schema

Browse source
This commit is contained in:
Meta Vens 2026-06-27 21:22:25 +01:00
commit 8b58208dff
3 changed files with 90 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

32
src/GameCult.Networking/CultNetDatabase.cs
View file

@ -593,7 +593,7 @@ namespace GameCult.Networking
foreach (var document in response.Documents.Where(document => ShardMatchesRawDocument(shard, document)))
{
var key = new CultRecordKey(document.RecordKey);
var descriptor = _cache.Registry.GetRequiredBySchemaId(document.SchemaId);
var descriptor = _documents.ResolveDescriptorForRawDocument(document);
var previous = _cache.Get(key);
var applied = await _documents.ApplyRawDocumentPutMessageAsync(
_cache,
@ -608,7 +608,7 @@ namespace GameCult.Networking
descriptor.DocumentType,
previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated,
key,
document.SchemaId,
descriptor.SchemaId,
shard,
applied,
previous);
@ -830,9 +830,9 @@ namespace GameCult.Networking
}
var key = new CultRecordKey(message.Document.RecordKey);
var shard = ResolveShardInternal(message.Document.SchemaId, key);
EnsurePrimary(shard, message.Document.SchemaId, key, message.ShardEpoch);
var descriptor = _cache.Registry.GetRequiredBySchemaId(message.Document.SchemaId);
var descriptor = _documents.ResolveDescriptorForRawDocument(message.Document);
var shard = ResolveShardInternal(descriptor.SchemaId, key);
EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch);
var previous = _cache.Get(key);
var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false);
var kind = previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated;
@ -1097,10 +1097,22 @@ namespace GameCult.Networking
private bool ShardMatchesRawDocument(CultNetShardDescriptor shard, CultNetRawDocumentRecord document)
{
return document != null &&
!string.IsNullOrWhiteSpace(document.SchemaId) &&
!string.IsNullOrWhiteSpace(document.RecordKey) &&
shard.Matches(document.SchemaId, new CultRecordKey(document.RecordKey));
if (document == null ||
string.IsNullOrWhiteSpace(document.SchemaId) ||
string.IsNullOrWhiteSpace(document.RecordKey))
{
return false;
}
try
{
var descriptor = _documents.ResolveDescriptorForRawDocument(document);
return shard.Matches(descriptor.SchemaId, new CultRecordKey(document.RecordKey));
}
catch (InvalidOperationException)
{
return false;
}
}
private void RemoveTrackedDocument(Type documentType, CultRecordKey key)
@ -1249,7 +1261,7 @@ namespace GameCult.Networking
}
var key = new CultRecordKey(message.Document.RecordKey);
var descriptor = _cache.Registry.GetRequiredBySchemaId(message.Document.SchemaId);
var descriptor = _documents.ResolveDescriptorForRawDocument(message.Document);
var previous = _cache.Get(key);
var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false);
var kind = ChangeKindFromWire(entry.ChangeKind);

2
src/GameCult.Networking/CultNetDocumentRegistry.cs
View file

@ -395,7 +395,7 @@ namespace GameCult.Networking
return applied;
}
private CultDocumentDescriptor ResolveDescriptorForRawDocument(CultNetRawDocumentRecord document)
internal CultDocumentDescriptor ResolveDescriptorForRawDocument(CultNetRawDocumentRecord document)
{
try
{

67
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -1750,6 +1750,59 @@ namespace GameCult.Networking.Tests
Assert.That(changes[0].Document!.Username, Is.EqualTo("Raw"));
}
[Test]
public async Task CultNetDatabase_ApplyRawPut_ResolvesForeignSchemaIdFromPayload()
{
var cache = new CultCache();
var registry = new CultNetDocumentRegistry(cache.Registry)
.Register(CultNetDocumentBinding.ForDocument<NetworkSchemaNote>(cache.Registry));
var descriptor = cache.Registry.GetRequired<NetworkSchemaNote>();
var shard = new CultNetShardDescriptor(
"network-notes",
"runtime-a",
epoch: 3,
isPrimary: true,
schemaIds: [descriptor.SchemaId],
keyPrefix: "network-note:",
primaryEndpoints: ["cultnet://runtime-a:3075"]);
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
{
DocumentRegistry = registry,
Shards = [shard]
});
var key = new CultRecordKey("network-note:foreign-schema-put");
var changes = new List<CultNetDatabaseChange<NetworkSchemaNote>>();
using var subscription = database.WatchRecord<NetworkSchemaNote>(key)
.Subscribe(change => changes.Add(change));
var message = new CultNetDocumentPutRawMessage
{
MessageId = "put-foreign-schema",
ShardId = shard.ShardId,
ShardEpoch = shard.Epoch,
Document = new CultNetRawDocumentRecord
{
SchemaId = "runtime.generated.network-note.ui.42",
RecordKey = key.Value,
StoredAt = DateTimeOffset.UtcNow.ToString("O"),
PayloadEncoding = "messagepack",
Payload = CultDocumentMessagePackSerialization.Serialize(new NetworkSchemaNote
{
Schema = "tests.networking_note.v1",
Text = "payload-routed",
Revision = 5
})
}
};
var applied = await database.ApplyPutAsync(message);
Assert.That(applied, Is.TypeOf<NetworkSchemaNote>());
Assert.That(cache.Get<NetworkSchemaNote>(key)!.Text, Is.EqualTo("payload-routed"));
Assert.That(changes, Has.Count.EqualTo(1));
Assert.That(changes[0].SchemaId, Is.EqualTo(descriptor.SchemaId));
Assert.That(changes[0].Document!.Revision, Is.EqualTo(5));
}
[Test]
public async Task CultNetDatabaseServer_Creates_Filtered_SnapshotResponse()
{
@ -4126,6 +4179,20 @@ namespace GameCult.Networking.Tests
public string Body { get; set; } = string.Empty;
}
[CultDocument("tests.networking_note", "tests.networking_note.v1")]
[MessagePack.MessagePackObject]
public sealed class NetworkSchemaNote
{
[MessagePack.Key(0)]
public string Schema { get; set; } = string.Empty;
[MessagePack.Key(1)]
public string Text { get; set; } = string.Empty;
[MessagePack.Key(2)]
public int Revision { get; set; }
}
private sealed class EnvironmentVariableScope : IDisposable
{
private readonly (string Name, string? Value)[] _originalValues;