Fix #8302 redis subscriptions with union payload fields

src/HotChocolate/Core/src/Subscriptions/DefaultJsonMessageSerializer.cs

@@ -14,7 +14,7 @@ public sealed class DefaultJsonMessageSerializer : IMessageSerializer
     private const string Completed = "{\"kind\":1}";
 
     private static readonly JsonSerializerOptions s_options =
-        JsonSerializerOptionDefaults.GraphQL;
+        CreateOptions();
 
     /// <inheritdoc />
     public string CompleteMessage => Completed;
@@ -56,4 +56,11 @@ private struct InternalMessageEnvelope<TBody>
         /// </summary>
         public MessageKind Kind { get; set; }
     }
+
+    private static JsonSerializerOptions CreateOptions()
+    {
+        var options = new JsonSerializerOptions(JsonSerializerOptionDefaults.GraphQL);
+        options.Converters.Add(new PolymorphicMessageJsonConverterFactory());
+        return options;
+    }
 }

src/HotChocolate/Core/src/Subscriptions/PolymorphicMessageJsonConverterFactory.cs

@@ -0,0 +1,109 @@
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace HotChocolate.Subscriptions;
+
+[UnconditionalSuppressMessage(
+    "Aot",
+    "IL3050",
+    Justification = "Subscription message serialization is runtime-based by design and already guarded on the public API.")]
+[UnconditionalSuppressMessage(
+    "Trimming",
+    "IL2026",
+    Justification = "Subscription message serialization is runtime-based by design and already guarded on the public API.")]
+[UnconditionalSuppressMessage(
+    "Trimming",
+    "IL2057",
+    Justification = "Runtime type names are produced by the serializer and validated to be assignable before use.")]
+internal sealed class PolymorphicMessageJsonConverterFactory : JsonConverterFactory
+{
+    public override bool CanConvert(Type typeToConvert)
+        => typeToConvert.IsInterface || typeToConvert.IsAbstract;
+
+    public override JsonConverter CreateConverter(Type typeToConvert, JsonSerializerOptions options)
+    {
+        var converterType = typeof(PolymorphicMessageJsonConverter<>).MakeGenericType(typeToConvert);
+        return (JsonConverter)Activator.CreateInstance(converterType)!;
+    }
+
+    private sealed class PolymorphicMessageJsonConverter<TValue> : JsonConverter<TValue>
+    {
+        private const string TypePropertyName = "$type";
+        private const string ValuePropertyName = "$value";
+
+        public override TValue? Read(
+            ref Utf8JsonReader reader,
+            Type typeToConvert,
+            JsonSerializerOptions options)
+        {
+            if (reader.TokenType is JsonTokenType.Null)
+            {
+                return default;
+            }
+
+            using var document = JsonDocument.ParseValue(ref reader);
+            var root = document.RootElement;
+
+            if (root.ValueKind is not JsonValueKind.Object
+                || !root.TryGetProperty(TypePropertyName, out var typeProperty)
+                || !root.TryGetProperty(ValuePropertyName, out var valueProperty))
+            {
+                throw new JsonException(
+                    $"Polymorphic value for '{typeToConvert.FullName}' must contain '{TypePropertyName}' and '{ValuePropertyName}'.");
+            }
+
+            var typeName = typeProperty.GetString();
+            if (string.IsNullOrEmpty(typeName))
+            {
+                throw new JsonException(
+                    $"Polymorphic value for '{typeToConvert.FullName}' did not specify a runtime type.");
+            }
+
+            var runtimeType = Type.GetType(typeName, throwOnError: false);
+            if (runtimeType is null || !typeToConvert.IsAssignableFrom(runtimeType))
+            {
+                throw new JsonException(
+                    $"Runtime type '{typeName}' is not assignable to '{typeToConvert.FullName}'.");
+            }
+
+            if (runtimeType.IsInterface || runtimeType.IsAbstract)
+            {
+                throw new JsonException(
+                    $"Runtime type '{runtimeType.FullName}' must be concrete.");
+            }
+
+            var value = JsonSerializer.Deserialize(
+                valueProperty.GetRawText(),
+                runtimeType,
+                options);
+
+            return (TValue?)value;
+        }
+
+        public override void Write(
+            Utf8JsonWriter writer,
+            TValue value,
+            JsonSerializerOptions options)
+        {
+            if (value is null)
+            {
+                writer.WriteNullValue();
+                return;
+            }
+
+            var runtimeType = value.GetType();
+            if (runtimeType.IsInterface || runtimeType.IsAbstract)
+            {
+                throw new JsonException(
+                    $"Runtime type '{runtimeType.FullName}' must be concrete.");
+            }
+
+            writer.WriteStartObject();
+            writer.WriteString(TypePropertyName, runtimeType.AssemblyQualifiedName);
+            writer.WritePropertyName(ValuePropertyName);
+            JsonSerializer.Serialize(writer, value, runtimeType, options);
+            writer.WriteEndObject();
+        }
+    }
+}

src/HotChocolate/Core/test/Subscriptions.Redis.Tests/RedisIntegrationTests.cs

@@ -1,5 +1,6 @@
 using HotChocolate.Execution;
 using HotChocolate.Execution.Configuration;
+using HotChocolate.Types;
 using Microsoft.Extensions.DependencyInjection;
 using Squadron;
 using StackExchange.Redis;
@@ -49,6 +50,67 @@ public override Task Subscribe_And_Complete_Topic()
     public override Task Subscribe_And_Complete_Topic_With_ValueTypeMessage()
         => base.Subscribe_And_Complete_Topic_With_ValueTypeMessage();
 
+    [Fact]
+    public async Task Subscribe_Union_Field_In_Payload()
+    {
+        using var cts = new CancellationTokenSource(Timeout);
+        await using var services = CreateServer(
+            builder => builder
+                .AddSubscriptionType<UnionPayloadSubscription>()
+                .AddType<UnionTextMessage>()
+                .AddType<UnionCodeMessage>()
+                .ModifyOptions(o => o.StrictValidation = false));
+        var sender = services.GetRequiredService<ITopicEventSender>();
+
+        var result = await services.ExecuteRequestAsync(
+            """
+            subscription {
+              onUnionPayload {
+                message {
+                  __typename
+                  ... on UnionTextMessage {
+                    text
+                  }
+                }
+              }
+            }
+            """,
+            cancellationToken: cts.Token);
+
+        await using var responseStream = result.ExpectResponseStream();
+        var responses = responseStream.ReadResultsAsync().ConfigureAwait(false);
+
+        await sender.SendAsync(
+            "OnUnionPayload",
+            new UnionPayloadEnvelope
+            {
+                Message = new UnionTextMessage { Text = "from-redis" }
+            },
+            cts.Token);
+        await sender.CompleteAsync("OnUnionPayload");
+
+        var snapshot = new Snapshot();
+
+        await foreach (var response in responses.WithCancellation(cts.Token).ConfigureAwait(false))
+        {
+            snapshot.Add(response);
+        }
+
+        snapshot.MatchInline(
+            """
+            {
+              "data": {
+                "onUnionPayload": {
+                  "message": {
+                    "__typename": "UnionTextMessage",
+                    "text": "from-redis"
+                  }
+                }
+              }
+            }
+            """);
+    }
+
     [Fact]
     public async Task Unsubscribe_Should_RemoveChannel()
     {
@@ -103,4 +165,30 @@ private async Task<RedisResult[]> GetActiveChannelsAsync()
 
     protected override void ConfigurePubSub(IRequestExecutorBuilder graphqlBuilder)
         => graphqlBuilder.AddRedisSubscriptions(_ => _redisResource.GetConnection());
+
+    [UnionType]
+    public interface IUnionMessage;
+
+    public sealed class UnionTextMessage : IUnionMessage
+    {
+        public string Text { get; set; } = default!;
+    }
+
+    public sealed class UnionCodeMessage : IUnionMessage
+    {
+        public int Code { get; set; }
+    }
+
+    public sealed class UnionPayloadEnvelope
+    {
+        public IUnionMessage Message { get; set; } = default!;
+    }
+
+    public sealed class UnionPayloadSubscription
+    {
+        [Topic("OnUnionPayload")]
+        [Subscribe]
+        public UnionPayloadEnvelope OnUnionPayload([EventMessage] UnionPayloadEnvelope message)
+            => message;
+    }
 }

src/HotChocolate/Core/test/Subscriptions.Tests/DefaultJsonMessageSerializerTests.cs

@@ -76,8 +76,42 @@ public void SerializeDefaultMessage()
             .MatchInline("{\"body\":\"abc\",\"kind\":0}");
     }
 
+    [Fact]
+    public void Serialize_And_Deserialize_DefaultMessage_With_Interface_Property()
+    {
+        // arrange
+        var serializer = new DefaultJsonMessageSerializer();
+        var message = new ContainerMessage
+        {
+            Payload = new TextPayload
+            {
+                Text = "abc"
+            }
+        };
+
+        // act
+        var serialized = serializer.Serialize(message);
+        var envelope = serializer.Deserialize<ContainerMessage>(serialized);
+
+        // assert
+        var payload = Assert.IsType<TextPayload>(envelope.Body!.Payload);
+        Assert.Equal("abc", payload.Text);
+    }
+
     public enum Foo
     {
         Bar
     }
+
+    public interface IPayload;
+
+    public sealed class TextPayload : IPayload
+    {
+        public string Text { get; set; } = default!;
+    }
+
+    public sealed class ContainerMessage
+    {
+        public IPayload Payload { get; set; } = default!;
+    }
 }