Add Fusion Subscriptions

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/FusionBuiltIns.cs

@@ -28,7 +28,8 @@ internal static class FusionBuiltIns
             new OverrideMutableDirectiveDefinition(s_stringType),
             new ProvidesMutableDirectiveDefinition(s_fieldSelectionSetType),
             new RequireMutableDirectiveDefinition(s_fieldSelectionMapType),
-            new ShareableMutableDirectiveDefinition()
+            new ShareableMutableDirectiveDefinition(),
+            new SubscribeMutableDirectiveDefinition(s_fieldSelectionSetType, s_stringType)
         ]).ToFrozenDictionary(d => d.Name);
 
     public static FrozenDictionary<string, MutableScalarTypeDefinition> SourceSchemaScalars { get; } =

src/HotChocolate/Fusion/src/Fusion.Composition/Info/SubscribeDirectiveInfo.cs

@@ -0,0 +1,9 @@
+using System.Collections.Immutable;
+using HotChocolate.Language;
+
+namespace HotChocolate.Fusion.Info;
+
+internal readonly record struct SubscribeDirectiveInfo(
+    ImmutableArray<string> Topics,
+    string? Broker,
+    SelectionSetNode Message);

src/HotChocolate/Fusion/src/Fusion.Composition/Logging/LogEntryCodes.cs

@@ -47,6 +47,7 @@ public static class LogEntryCodes
     public const string LookupMustHaveArguments = "LOOKUP_MUST_HAVE_ARGUMENTS";
     public const string LookupReturnsList = "LOOKUP_RETURNS_LIST";
     public const string LookupReturnsNonNullableType = "LOOKUP_RETURNS_NON_NULLABLE_TYPE";
+    public const string MultipleSubscribeSources = "MULTIPLE_SUBSCRIBE_SOURCES";
     public const string NonNullInputFieldIsInaccessible = "NON_NULL_INPUT_FIELD_IS_INACCESSIBLE";
     public const string NoQueries = "NO_QUERIES";
     public const string OutputFieldTypesNotMergeable = "OUTPUT_FIELD_TYPES_NOT_MERGEABLE";

src/HotChocolate/Fusion/src/Fusion.Composition/Logging/LogEntryHelper.cs

@@ -569,6 +569,22 @@ public static LogEntry InvalidShareableUsage(
             .Build();
     }
 
+    public static LogEntry MultipleSubscribeSources(
+        MutableOutputFieldDefinition field,
+        MutableSchemaDefinition schema)
+    {
+        return LogEntryBuilder.New()
+            .SetMessage(
+                LogEntryHelper_MultipleSubscribeSources,
+                field.Coordinate.ToString(),
+                schema.Name)
+            .SetCode(LogEntryCodes.MultipleSubscribeSources)
+            .SetSeverity(LogSeverity.Error)
+            .SetTypeSystemMember(field)
+            .SetSchema(schema)
+            .Build();
+    }
+
     public static LogEntry IsInvalidFields(
         Directive isDirective,
         MutableInputFieldDefinition argument,

src/HotChocolate/Fusion/src/Fusion.Composition/PreMergeValidationRules/MultipleSubscribeSourcesRule.cs

@@ -0,0 +1,174 @@
+using System.Collections.Immutable;
+using System.Text;
+using HotChocolate.Fusion.Events;
+using HotChocolate.Fusion.Events.Contracts;
+using HotChocolate.Fusion.Extensions;
+using HotChocolate.Fusion.Info;
+using HotChocolate.Language;
+using HotChocolate.Types.Mutable;
+using static HotChocolate.Fusion.Logging.LogEntryHelper;
+
+namespace HotChocolate.Fusion.PreMergeValidationRules;
+
+internal sealed class MultipleSubscribeSourcesRule : IEventHandler<OutputFieldGroupEvent>
+{
+    public void Handle(OutputFieldGroupEvent @event, CompositionContext context)
+    {
+        var fieldGroup = @event.FieldGroup;
+        ImmutableArray<SubscribeContribution>.Builder? builder = null;
+
+        foreach (var (field, _, schema) in fieldGroup)
+        {
+            foreach (var directive in field.GetSubscribeDirectives())
+            {
+                builder ??= ImmutableArray.CreateBuilder<SubscribeContribution>();
+                builder.Add(new SubscribeContribution(field, schema, field.IsShareable, directive));
+            }
+        }
+
+        var contributions = builder?.ToImmutable() ?? [];
+
+        if (contributions.Length < 2 || contributions.Any(t => !t.IsShareable))
+        {
+            return;
+        }
+
+        for (var i = 1; i < contributions.Length; i++)
+        {
+            if (!TypeMergeHelper.SameTypeShape(
+                contributions[0].Field.Type,
+                contributions[i].Field.Type))
+            {
+                context.Log.Write(
+                    OutputFieldTypesNotMergeable(
+                        contributions[0].Field,
+                        contributions[0].Schema,
+                        contributions[i].Schema));
+                return;
+            }
+        }
+
+        var reference = SubscribeIdentity.Create(contributions[0].Directive);
+
+        for (var i = 1; i < contributions.Length; i++)
+        {
+            if (!SubscribeIdentity.Create(contributions[i].Directive).Equals(reference))
+            {
+                context.Log.Write(MultipleSubscribeSources(contributions[0].Field, contributions[0].Schema));
+                return;
+            }
+        }
+    }
+
+    private readonly record struct SubscribeContribution(
+        MutableOutputFieldDefinition Field,
+        MutableSchemaDefinition Schema,
+        bool IsShareable,
+        SubscribeDirectiveInfo Directive);
+
+    private readonly record struct SubscribeIdentity(
+        string? Broker,
+        string Topics,
+        string Message)
+    {
+        public static SubscribeIdentity Create(SubscribeDirectiveInfo directive)
+        {
+            return new SubscribeIdentity(
+                directive.Broker,
+                NormalizeTopics(directive.Topics),
+                NormalizeSelectionSet(directive.Message));
+        }
+    }
+
+    private static string NormalizeTopics(ImmutableArray<string> topics)
+    {
+        if (topics.IsDefaultOrEmpty)
+        {
+            return "";
+        }
+
+        var builder = new StringBuilder();
+        var first = true;
+
+        foreach (var topic in topics
+            .Distinct(StringComparer.Ordinal)
+            .Order(StringComparer.Ordinal))
+        {
+            if (!first)
+            {
+                builder.Append('\0');
+            }
+
+            builder.Append(topic);
+            first = false;
+        }
+
+        return builder.ToString();
+    }
+
+    private static string NormalizeSelectionSet(SelectionSetNode selectionSet)
+    {
+        var builder = new StringBuilder();
+        AppendSelectionSet(builder, selectionSet);
+        return builder.ToString();
+    }
+
+    private static void AppendSelectionSet(StringBuilder builder, SelectionSetNode selectionSet)
+    {
+        var selections = selectionSet.Selections
+            .Select(FormatSelection)
+            .Order(StringComparer.Ordinal);
+        var first = true;
+
+        foreach (var selection in selections)
+        {
+            if (!first)
+            {
+                builder.Append(' ');
+            }
+
+            builder.Append(selection);
+            first = false;
+        }
+    }
+
+    private static string FormatSelection(ISelectionNode selection)
+    {
+        var builder = new StringBuilder();
+
+        switch (selection)
+        {
+            case FieldNode field:
+                if (field.Alias is not null)
+                {
+                    builder.Append(field.Alias.Value).Append(':');
+                }
+
+                builder.Append(field.Name.Value);
+
+                if (field.SelectionSet is not null)
+                {
+                    builder.Append('{');
+                    AppendSelectionSet(builder, field.SelectionSet);
+                    builder.Append('}');
+                }
+
+                break;
+
+            case InlineFragmentNode inlineFragment:
+                builder.Append("...on ");
+                builder.Append(inlineFragment.TypeCondition?.Name.Value);
+                builder.Append('{');
+                AppendSelectionSet(builder, inlineFragment.SelectionSet);
+                builder.Append('}');
+                break;
+
+            case FragmentSpreadNode fragmentSpread:
+                builder.Append("...");
+                builder.Append(fragmentSpread.Name.Value);
+                break;
+        }
+
+        return builder.ToString();
+    }
+}