Add Fusion Subscriptions

src/All.slnx

@@ -221,6 +221,8 @@
     <Project Path="HotChocolate/Fusion/src/Fusion.Execution/HotChocolate.Fusion.Execution.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.Language/HotChocolate.Fusion.Language.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.Packaging/HotChocolate.Fusion.Packaging.csproj" />
+    <Project Path="HotChocolate/Fusion/src/Fusion.Subscriptions.Kafka/HotChocolate.Fusion.Subscriptions.Kafka.csproj" />
+    <Project Path="HotChocolate/Fusion/src/Fusion.Subscriptions.NATS/HotChocolate.Fusion.Subscriptions.NATS.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.Utilities/HotChocolate.Fusion.Utilities.csproj" />
     <Project Path="HotChocolate/Fusion/src/Fusion.SourceSchema.Packaging/HotChocolate.Fusion.SourceSchema.Packaging.csproj" />
   </Folder>
@@ -231,6 +233,8 @@
     <Project Path="HotChocolate/Fusion/test/Fusion.Connectors.ApolloFederation.Compliance.Tests/HotChocolate.Fusion.Connectors.ApolloFederation.Compliance.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Connectors.ApolloFederation.Tests/HotChocolate.Fusion.Connectors.ApolloFederation.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Connectors.InMemory.Tests/HotChocolate.Fusion.Connectors.InMemory.Tests.csproj" />
+    <Project Path="HotChocolate/Fusion/test/Fusion.Subscriptions.Kafka.Tests/HotChocolate.Fusion.Subscriptions.Kafka.Tests.csproj" />
+    <Project Path="HotChocolate/Fusion/test/Fusion.Subscriptions.NATS.Tests/HotChocolate.Fusion.Subscriptions.NATS.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Caching.Tests/HotChocolate.Fusion.Caching.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Diagnostics.Tests/HotChocolate.Fusion.Diagnostics.Tests.csproj" />
     <Project Path="HotChocolate/Fusion/test/Fusion.Execution.Tests/HotChocolate.Fusion.Execution.Tests.csproj" />

src/Directory.Packages.props

@@ -15,6 +15,7 @@
     <PackageVersion Include="BenchmarkDotNet.Diagnostics.dotMemory" Version="0.15.8" />
     <PackageVersion Include="ChilliCream.ModelContextProtocol.AspNetCore" Version="1.4.0" />
     <PackageVersion Include="ChilliCream.Nitro.App" Version="$(NitroVersion)" />
+    <PackageVersion Include="Confluent.Kafka" Version="2.14.2" />
     <PackageVersion Include="DiffPlex" Version="1.9.0" />
     <PackageVersion Include="Duende.IdentityModel" Version="7.1.0" />
     <PackageVersion Include="Duende.IdentityModel.OidcClient" Version="6.0.1" />
@@ -39,6 +40,7 @@
     <PackageVersion Include="MongoDB.Driver" Version="3.0.0" />
     <PackageVersion Include="Moq" Version="4.20.72" />
     <PackageVersion Include="NATS.Client.Core" Version="2.7.2" />
+    <PackageVersion Include="NATS.Client.JetStream" Version="2.7.2" />
     <PackageVersion Include="NATS.Extensions.Microsoft.DependencyInjection" Version="2.7.2" />
     <PackageVersion Include="NetTopologySuite" Version="2.0.0" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.2" />

src/HotChocolate/Fusion/HotChocolate.Fusion.slnx

@@ -15,6 +15,8 @@
     <Project Path="src/Fusion.Packaging/HotChocolate.Fusion.Packaging.csproj" />
     <Project Path="src/Fusion.SourceSchema.Packaging/HotChocolate.Fusion.SourceSchema.Packaging.csproj" />
     <Project Path="src/Fusion.Connectors.InMemory/HotChocolate.Fusion.Connectors.InMemory.csproj" />
+    <Project Path="src/Fusion.Subscriptions.Kafka/HotChocolate.Fusion.Subscriptions.Kafka.csproj" />
+    <Project Path="src/Fusion.Subscriptions.NATS/HotChocolate.Fusion.Subscriptions.NATS.csproj" />
     <Project Path="src/Fusion.Utilities/HotChocolate.Fusion.Utilities.csproj" />
   </Folder>
   <Folder Name="/test/">
@@ -30,6 +32,8 @@
     <Project Path="test/Fusion.Packaging.Tests/HotChocolate.Fusion.Packaging.Tests.csproj" />
     <Project Path="test/Fusion.SourceSchema.Packaging.Tests/HotChocolate.Fusion.SourceSchema.Packaging.Tests.csproj" />
     <Project Path="test/Fusion.Connectors.InMemory.Tests/HotChocolate.Fusion.Connectors.InMemory.Tests.csproj" />
+    <Project Path="test/Fusion.Subscriptions.Kafka.Tests/HotChocolate.Fusion.Subscriptions.Kafka.Tests.csproj" />
+    <Project Path="test/Fusion.Subscriptions.NATS.Tests/HotChocolate.Fusion.Subscriptions.NATS.Tests.csproj" />
     <Project Path="test/Fusion.Utilities.Tests/HotChocolate.Fusion.Utilities.Tests.csproj" />
     <Project Path="test/Fusion.Tests.Shared/HotChocolate.Fusion.Tests.Shared.csproj" />
   </Folder>

src/HotChocolate/Fusion/src/Fusion.Composition/Collections/SatisfiabilityPath.cs

@@ -1,5 +1,6 @@
 using System.Collections;
 using System.Diagnostics.CodeAnalysis;
+using HotChocolate.Language;
 using HotChocolate.Types;
 using HotChocolate.Types.Mutable;
 
@@ -94,6 +95,8 @@ internal sealed record SatisfiabilityPathItem(
 {
     public ITypeDefinition FieldType { get; } = Field.Type.AsTypeDefinition();
 
+    public SelectionSetNode? ProvidedSelectionSet { get; init; }
+
     private readonly int _hashCode = HashCode.Combine(Field, Type, SchemaName);
 
     public override string ToString()

src/HotChocolate/Fusion/src/Fusion.Composition/CompositionHelper.cs

@@ -53,8 +53,7 @@ await archive.GetSourceSchemaNamesAsync(cancellationToken),
                         .SetSeverity(LogSeverity.Error)
                         .Build());
 
-                ImmutableArray<CompositionError> errors = [new("❌ Composition failed")];
-                return errors;
+                return (ImmutableArray<CompositionError>)[new("❌ Composition failed")];
             }
         }
 

src/HotChocolate/Fusion/src/Fusion.Composition/Definitions/FusionSubscribeMutableDirectiveDefinition.cs

@@ -0,0 +1,27 @@
+using HotChocolate.Types;
+using HotChocolate.Types.Mutable;
+using static HotChocolate.Fusion.WellKnownArgumentNames;
+using static HotChocolate.Fusion.WellKnownDirectiveNames;
+
+namespace HotChocolate.Fusion.Definitions;
+
+/// <summary>
+/// The <c>@fusion__subscribe</c> directive specifies broker metadata for a composed
+/// subscription field.
+/// </summary>
+internal sealed class FusionSubscribeMutableDirectiveDefinition : MutableDirectiveDefinition
+{
+    public FusionSubscribeMutableDirectiveDefinition(
+        MutableEnumTypeDefinition schemaMutableEnumType,
+        MutableScalarTypeDefinition fieldSelectionSetType,
+        MutableScalarTypeDefinition stringType)
+        : base(FusionSubscribe)
+    {
+        Arguments.Add(new MutableInputFieldDefinition(Schema, new NonNullType(schemaMutableEnumType)));
+        Arguments.Add(new MutableInputFieldDefinition(Topics, new ListType(new NonNullType(stringType))));
+        Arguments.Add(new MutableInputFieldDefinition(Broker, stringType));
+        Arguments.Add(new MutableInputFieldDefinition(Message, new NonNullType(fieldSelectionSetType)));
+
+        Locations = DirectiveLocation.FieldDefinition;
+    }
+}

src/HotChocolate/Fusion/src/Fusion.Composition/Definitions/SubscribeMutableDirectiveDefinition.cs

@@ -0,0 +1,31 @@
+using HotChocolate.Types;
+using HotChocolate.Types.Mutable;
+
+namespace HotChocolate.Fusion.Definitions;
+
+/// <summary>
+/// The <c>@subscribe</c> directive declares event-stream metadata for a subscription field.
+/// </summary>
+internal sealed class SubscribeMutableDirectiveDefinition : MutableDirectiveDefinition
+{
+    public SubscribeMutableDirectiveDefinition(
+        MutableScalarTypeDefinition fieldSelectionSetType,
+        MutableScalarTypeDefinition stringType)
+        : base(WellKnownDirectiveNames.Subscribe)
+    {
+        Arguments.Add(
+            new MutableInputFieldDefinition(
+                WellKnownArgumentNames.Topics,
+                new ListType(new NonNullType(stringType))));
+        Arguments.Add(
+            new MutableInputFieldDefinition(
+                WellKnownArgumentNames.Broker,
+                stringType));
+        Arguments.Add(
+            new MutableInputFieldDefinition(
+                WellKnownArgumentNames.Message,
+                new NonNullType(fieldSelectionSetType)));
+
+        Locations = DirectiveLocation.FieldDefinition;
+    }
+}

src/HotChocolate/Fusion/src/Fusion.Composition/Extensions/DirectivesProviderExtensions.cs

@@ -1,3 +1,5 @@
+using System.Collections.Immutable;
+using HotChocolate.Fusion.Info;
 using HotChocolate.Language;
 using HotChocolate.Types;
 using HotChocolate.Types.Mutable;
@@ -55,6 +57,34 @@ public void AddDirective(Directive directive)
             return null;
         }
 
+        public ImmutableArray<SubscribeDirectiveInfo> GetSubscribeDirectives()
+        {
+            ImmutableArray<SubscribeDirectiveInfo>.Builder? builder = null;
+
+            foreach (var subscribeDirective in member.Directives.AsEnumerable())
+            {
+                if (subscribeDirective.Name != WellKnownDirectiveNames.Subscribe)
+                {
+                    continue;
+                }
+
+                if (!subscribeDirective.Arguments.TryGetValue(ArgumentNames.Message, out var message)
+                    || message is not StringValueNode messageArgument)
+                {
+                    continue;
+                }
+
+                builder ??= ImmutableArray.CreateBuilder<SubscribeDirectiveInfo>();
+                builder.Add(
+                    new SubscribeDirectiveInfo(
+                        GetOptionalStringListArgument(subscribeDirective, ArgumentNames.Topics),
+                        GetOptionalStringArgument(subscribeDirective, ArgumentNames.Broker),
+                        ParseSelectionSet(messageArgument.Value)));
+            }
+
+            return builder?.ToImmutable() ?? [];
+        }
+
         public HashSet<string> GetTags()
         {
             var tags = new HashSet<string>();
@@ -89,4 +119,44 @@ public bool HasInaccessibleDirective()
             return member.Directives.ContainsName(WellKnownDirectiveNames.Inaccessible);
         }
     }
+
+    private static string? GetOptionalStringArgument(IDirective directive, string name)
+        => directive.Arguments.TryGetValue(name, out var value) && value is StringValueNode stringValue
+            ? stringValue.Value
+            : null;
+
+    private static ImmutableArray<string> GetOptionalStringListArgument(
+        IDirective directive,
+        string name)
+    {
+        if (!directive.Arguments.TryGetValue(name, out var value)
+            || value is not ListValueNode listValue)
+        {
+            return [];
+        }
+
+        var builder = ImmutableArray.CreateBuilder<string>(listValue.Items.Count);
+
+        foreach (var item in listValue.Items)
+        {
+            if (item is StringValueNode stringValue)
+            {
+                builder.Add(stringValue.Value);
+            }
+        }
+
+        return builder.ToImmutable();
+    }
+
+    private static SelectionSetNode ParseSelectionSet(string value)
+    {
+        try
+        {
+            return Utf8GraphQLParser.Syntax.ParseSelectionSet(value);
+        }
+        catch (SyntaxException)
+        {
+            return Utf8GraphQLParser.Syntax.ParseSelectionSet($"{{ {value} }}");
+        }
+    }
 }

src/HotChocolate/Fusion/src/Fusion.Composition/Extensions/MutableOutputFieldDefinitionExtensions.cs

@@ -60,6 +60,40 @@ public bool ExistsInSchema(string schemaName)
             return null;
         }
 
+        public ImmutableArray<string> GetFusionSubscribeSchemaNames()
+        {
+            ImmutableArray<string>.Builder? builder = null;
+
+            foreach (var directive in field.Directives.AsEnumerable())
+            {
+                if (directive.Name != DirectiveNames.FusionSubscribe)
+                {
+                    continue;
+                }
+
+                builder ??= ImmutableArray.CreateBuilder<string>();
+                builder.Add((string)directive.Arguments[ArgumentNames.Schema].Value!);
+            }
+
+            return builder?.ToImmutable() ?? [];
+        }
+
+        public SelectionSetNode? GetFusionSubscribeMessage(string schemaName)
+        {
+            var fusionSubscribeDirective =
+                field.Directives.AsEnumerable().FirstOrDefault(
+                    d =>
+                        d.Name == DirectiveNames.FusionSubscribe
+                        && (string)d.Arguments[ArgumentNames.Schema].Value! == schemaName);
+
+            if (fusionSubscribeDirective?.Arguments.TryGetValue(ArgumentNames.Message, out var message) == true)
+            {
+                return ParseSelectionSet((string)message.Value!);
+            }
+
+            return null;
+        }
+
         public List<string> GetFusionLookupMap()
         {
             var items = new List<string>();