C# InteropPeer: receipt-serve (CultNetDatabaseServer on an ephemeral port) and receipt-relay (LiteNetLib byte shim for the TS client)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

tests/GameCult.Networking.InteropPeer/Program.cs

@@ -25,7 +25,7 @@ static async Task<int> ProgramMainAsync(string[] args)
     {
         if (args.Length == 0)
         {
-            throw new InvalidOperationException("Expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once | quic-realtime-serve");
+            throw new InvalidOperationException("Expected mode: serve | probe | dial | rudp-serve-once | rudp-dial-once | rudp-serve-message-once | rudp-dial-message-once | quic-realtime-serve | receipt-serve | receipt-relay");
         }
 
         var mode = args[0];
@@ -53,6 +53,12 @@ static async Task<int> ProgramMainAsync(string[] args)
             case "rudp-dial-message-once":
                 RudpDialMessageOnce(options);
                 return 0;
+            case "receipt-serve":
+                await GameCult.Networking.InteropPeer.ReceiptHarness.ServeAsync();
+                return 0;
+            case "receipt-relay":
+                await GameCult.Networking.InteropPeer.ReceiptHarness.RelayAsync(int.Parse(options["port"], CultureInfo.InvariantCulture));
+                return 0;
             case "quic-realtime-serve":
                 await QuicRealtimeServeAsync(ParseQuicRealtimeServeConfig(options));
                 return 0;

tests/GameCult.Networking.InteropPeer/ReceiptHarness.cs

@@ -0,0 +1,118 @@
+using System;
+using System.IO;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using GameCult.Caching;
+using GameCult.Networking;
+using LiteNetLib;
+using R3;
+
+// The real C# receipt server (CultNetDatabaseServer on Server) and the one transport shim the TypeScript client needs to reach it,
+// for the cross-runtime write-receipt interop harness (packages/cultmesh-ts/test/interop/write-receipt-interop.test.ts).
+// TypeScript has no LiteNetLib client, so the shim moves bytes and nothing else: each CultNet frame is a LiteNetLib reliable
+// ordered payload, and on the pipe it is a 4-byte big-endian length followed by the frame.
+namespace GameCult.Networking.InteropPeer
+{
+    [CultDocument("receipt.interop_note", ReceiptHarness.SchemaId)]
+    [MessagePack.MessagePackObject]
+    public sealed class ReceiptInteropNote
+    {
+        [MessagePack.Key(0)]
+        public string Text { get; set; } = string.Empty;
+    }
+
+    internal static class ReceiptHarness
+    {
+        public const string SchemaId = "receipt.interop_note.v1";
+        public const string ShardId = "receipt-interop-shard";
+        // A record key with this prefix is refused by the authority: how the harness asks for a refused write.
+        public const string RefusedPrefix = "refuse:";
+
+        private sealed class Authority : ICultNetRemoteWriteAuthority
+        {
+            public CultNetRemoteWriteDecision Decide(CultNetRemoteWriteRequest request) =>
+                request.RecordKey.ToString().StartsWith(RefusedPrefix, StringComparison.Ordinal)
+                    ? CultNetRemoteWriteDecision.Refuse("denied")
+                    : CultNetRemoteWriteDecision.Grant();
+        }
+
+        private static void Emit(object fact)
+        {
+            lock (Console.Out) Console.Out.WriteLine(JsonSerializer.Serialize(fact));
+        }
+
+        // stdout: {"event":"listening","port":N} then {"event":"applied"|"removed","recordKey":"..."} per committed change; runs until stdin closes.
+        public static async Task ServeAsync()
+        {
+            var cache = new CultCache();
+            var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
+            {
+                Shards = [new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: true, schemaIds: [SchemaId])]
+            });
+            using var server = new Server(cache, ServerSecurityOptions.Development()) { Port = 0 };
+            using var databaseServer = new CultNetDatabaseServer(server, database, new CultNetDatabaseServerOptions
+            {
+                RemoteWriteAuthority = new Authority()
+            });
+            using var watch = cache.Watch<ReceiptInteropNote>().Subscribe(change =>
+                Emit(new { @event = change.Kind == CultCacheDocumentChangeKind.Removed ? "removed" : "applied", recordKey = change.Key.ToString() }));
+            server.Start();
+            Emit(new { @event = "listening", port = server.Port, shardId = ShardId, schemaId = SchemaId });
+            await Task.Run(() => Console.In.ReadToEnd());
+        }
+
+        // Connects to the server on `port` and relays frames between stdin/stdout and the connection; exits when either ends.
+        public static async Task RelayAsync(int port)
+        {
+            var listener = new EventBasedNetListener();
+            var connected = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var ended = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var output = Console.OpenStandardOutput();
+            var gate = new object();
+            listener.PeerConnectedEvent += _ => connected.TrySetResult(true);
+            listener.PeerDisconnectedEvent += (_, _) => ended.TrySetResult(true);
+            listener.NetworkReceiveEvent += (_, reader, _, _) =>
+            {
+                var frame = reader.GetRemainingBytes();
+                var header = new byte[] { (byte)(frame.Length >> 24), (byte)(frame.Length >> 16), (byte)(frame.Length >> 8), (byte)frame.Length };
+                lock (gate)
+                {
+                    output.Write(header);
+                    output.Write(frame);
+                    output.Flush();
+                }
+            };
+            var manager = new NetManager(listener) { UnsyncedEvents = true };
+            manager.Start(0);
+            var peer = manager.Connect("127.0.0.1", port, ServerSecurityOptions.Development().ConnectionKey);
+            if (await Task.WhenAny(connected.Task, Task.Delay(TimeSpan.FromSeconds(10))) != connected.Task)
+                throw new InvalidOperationException("the relay did not connect");
+            var input = Console.OpenStandardInput();
+            var pump = Task.Run(() =>
+            {
+                var header = new byte[4];
+                while (ReadExactly(input, header))
+                {
+                    var frame = new byte[(header[0] << 24) | (header[1] << 16) | (header[2] << 8) | header[3]];
+                    if (!ReadExactly(input, frame)) break;
+                    peer.Send(frame, DeliveryMethod.ReliableOrdered);
+                }
+            });
+            await Task.WhenAny(pump, ended.Task);
+            manager.Stop();
+        }
+
+        private static bool ReadExactly(Stream stream, byte[] buffer)
+        {
+            var read = 0;
+            while (read < buffer.Length)
+            {
+                var count = stream.Read(buffer, read, buffer.Length - read);
+                if (count <= 0) return false;
+                read += count;
+            }
+            return true;
+        }
+    }
+}