[WIP] Introduce IOperationMessageListener

[sungam3r]

There was IOperationMessageListener in v6 and some our code uses it. Now I need to migrate that code to the-latest-and-greatest v7. My main goal is to achieve the "old behavior" without inheriting from GraphQLHttpMiddleware on our side since it creates complexity (combinations) with our other code that inherits from GraphQLHttpMiddleware in other way. So I looked throught the code and tried to find the way to hook up into message execution in the most soft way.

[codecov-commenter]

Codecov Report

Merging #1010 (89e9ce7) into master (9ebcb06) will decrease coverage by 0.16%. The diff coverage is 76.47%.

:mega: This organization is not using Codecov’s GitHub App Integration. We recommend you install it so Codecov can continue to function properly for your repositories. Learn more

@@            Coverage Diff             @@
##           master    #1010      +/-   ##
==========================================
- Coverage   93.42%   93.26%   -0.16%     
==========================================
  Files          44       44              
  Lines        2175     2184       +9     
  Branches      366      372       +6     
==========================================
+ Hits         2032     2037       +5     
- Misses        102      104       +2     
- Partials       41       43       +2     

Impacted Files	Coverage Δ
...NetCore/WebSockets/GraphQLWs/SubscriptionServer.cs	97.56% <66.66%> (-1.61%)	:arrow_down:
...ets/SubscriptionsTransportWs/SubscriptionServer.cs	97.60% <66.66%> (-1.59%)	:arrow_down:
src/Transports.AspNetCore/GraphQLHttpMiddleware.cs	95.22% <100.00%> (+0.03%)	:arrow_up:

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

[Shane32]

This seems to have a very limited use case. It cannot be used to change the protocol or execution at all. It can't be used to support new message types. Essentially, all it can be used for is logging -- at best, it could disconnect the client.

Can you provide your use case for consideration?

[sungam3r]

internal class SubscriptionHeadersListener : IOperationMessageListener
{
    private readonly IServiceCallContextAccessor _serviceCallContextAccessor;
    private readonly ServiceCallContextHttpFactory _serviceCallContextHttpFactory;
    private readonly IGraphQLSerializer _serializer;
    private readonly ILogger? _logger;

    public SubscriptionHeadersListener(
        IServiceCallContextAccessor serviceCallContextAccessor,
        ServiceCallContextHttpFactory serviceCallContextHttpFactory,
        IGraphQLSerializer serializer,
        ILoggerFactory? loggerFactory = null)
    {
        _serviceCallContextAccessor = serviceCallContextAccessor ?? throw new ArgumentNullException(nameof(serviceCallContextAccessor));
        _serviceCallContextHttpFactory = serviceCallContextHttpFactory ?? throw new ArgumentNullException(nameof(serviceCallContextHttpFactory));
        _serializer = serializer;
        _logger = loggerFactory?.CreateLogger(nameof(SubscriptionHeadersListener));
    }

    public Task BeforeHandleAsync(MessageHandlingContext context)
    {
        if (context == null)
            return Task.CompletedTask;

        try
        {
            var message = context.Message;

            if (message.Payload != null)
            {
                if (message.Type == MessageType.GQL_CONNECTION_INIT)
                {
                    var deserialized = _serializer.ReadNode<Dictionary<string, string?>>(message.Payload);

                    ModifyServiceCallContext(context, deserialized);
                }
                else if (message.Type == MessageType.GQL_START)
                {
                    var deserialized = _serializer.ReadNode<Dictionary<string, object?>>(message.Payload);
                    if (deserialized != null)
                    {
                        bool headersPresent = deserialized.TryGetValue("headers", out object? headers) || deserialized.TryGetValue("Headers", out headers);
                        if (headersPresent)
                            ModifyServiceCallContext(context, _serializer.ReadNode<Dictionary<string, string?>>(headers));
                    }
                }
            }
        }
        catch (Exception e)
        {
            context.Terminated = true;

            _logger?.LogError(
                e,
                "Some error {@OperationMessage}",
                new OperationMessage
                {
                    Id = context.Message?.Id,
                    Type = context.Message?.Type,
                    Payload = context.Message?.Payload?.ToString(),
                });
        }

        return Task.CompletedTask;
    }

    private void ModifyServiceCallContext(MessageHandlingContext context, Dictionary<string, string?>? headers)
    {
        var scc = _serviceCallContextAccessor.ServiceCallContext;

        if (scc == null)
        {
            return;
        }

        var httpContext = new DefaultHttpContext();

        if (headers?.Count > 0)
        {
            foreach (var header in headers)
            {
                httpContext.Request.Headers.Add(header.Key, header.Value);
            }
        }

        var result = _serviceCallContextHttpFactory.CreateServiceCallContextAsync(httpContext);

        if (result.SCC != null)
        {
            scc.MessageProperties = result.SCC.MessageProperties;
            scc.CancellationToken = result.SCC.CancellationToken;

            scc.Items.Clear();
            foreach (var item in result.SCC.Items)
            {
                scc.Items.TryAdd(item.Key, item.Value);
            }
        }
        else
        {
            if (context != null)
                context.Terminated = true;

            _logger?.LogError(
                "Some error {@OperationMessage}: {ServiceCallContextCreationError}",
                new OperationMessage
                {
                    Id = context?.Message?.Id,
                    Type = context?.Message?.Type,
                    Payload = context?.Message?.Payload?.ToString(),
                },
                result.Error);
        }
    }

    public Task HandleAsync(MessageHandlingContext context) => Task.CompletedTask;

    public Task AfterHandleAsync(MessageHandlingContext context)
    {
        var scc = _serviceCallContextAccessor.ServiceCallContext;

        if (context.Message.Type == MessageType.GQL_START && scc != null)
        {
            scc.MessageProperties = null;
            scc.CancellationToken = default;
            scc.Items.Clear();
        }

        return Task.CompletedTask;
    }
}

+ extension method to plug in that class into DI:

    public static IServiceCollection AddSubscriptionHeadersServiceCallContextIntegration(this IServiceCollection services)
    {
        services
            .AddServiceCallContextAccessor()
            .AddTransient<IOperationMessageListener, SubscriptionHeadersListener>();

        return services;
    }

[sungam3r]

It cannot be used to change the protocol or execution at all. It can't be used to support new message types.

I do not need all that.

[sungam3r]

ServiceCallContext - is a generalization of HttpContext that works in any environment - HTTP, RabbitMQ, Kafka, NSB, whatever. I need to fill it up with headers.

[Shane32]

GraphQL.NET Server v7 already contains the necessary abstractions to implement this within the GQL_CONNECTION_INIT message (and the corresponding message for the newer protocol). In fact, there are two different interfaces that can be used.

// abstract service call context initialization service
internal interface IServiceCallContextInitializer
{
    void ModifyServiceCallContext(HttpContext context, Dictionary<string, string?>? deserializedPayload, CancellationToken cancellationToken);
}

// Technique 1:
//
// Add to GraphQLBuilder with:
//   .AddWebSocketAuthentication<MyWebSocketAuthenticationService>();
internal class MyWebSocketAuthenticationService : IWebSocketAuthenticationService
{
    private readonly IGraphQLSerializer _serializer;
    private readonly IServiceCallContextInitializer _serviceCallContextInitializer;

    public MyWebSocketAuthenticationService(IGraphQLSerializer serializer, IServiceCallContextInitializer serviceCallContextInitializer)
    {
        _serializer = serializer;
        _serviceCallContextInitializer = serviceCallContextInitializer;
    }

    public Task AuthenticateAsync(IWebSocketConnection connection, string subProtocol, OperationMessage operationMessage)
    {

        var deserialized = _serializer.ReadNode<Dictionary<string, string?>>(operationMessage.Payload);

        var cancellationToken = connection.RequestAborted; // also tied to application lifetime rather than HttpContext.RequestAborted

        _serviceCallContextInitializer.ModifyServiceCallContext(connection.HttpContext, deserialized, cancellationToken);

        // connection can be closed if necessary via connection.CloseAsync or connection.HttpContext.Abort

        return Task.CompletedTask;
    }
}

// Technique 2:
builder.AddUserContextBuilder((context, payload) =>
{
    var serializer = context.RequestServices.GetRequiredService<IGraphQLSerializer>();
    var deserialized = serializer.ReadNode<Dictionary<string, string?>>(payload);
    var initializer = context.RequestServices.GetRequiredService<IServiceCallContextInitializer>();
    initializer.ModifyServiceCallContext(context, deserialized, context.RequestAborted);
    var userContext = new Dictionary<string, object?>();
    return Task.FromResult(userContext);
});

See: https://github.com/graphql-dotnet/server#authentication-for-websocket-requests

[Shane32]

Neither technique will allow for monitoring of GQL_START as this violates protocol (as the GQL_START payload should be a GraphQL request message aka GraphQLRequest) and could cause concurrency issues where the HttpContext (or ServiceCallContext in your case) can be modified while it is in use on another thread. Any headers-like data should be under the Extensions property, which you could access with a document listener.

[sungam3r]

So IWebSocketAuthenticationService is a extension point itself to hook up explicitly into OnConnectionInitAsync - particular single event.

Neither technique will allow for monitoring of GQL_START as this violates protocol

I do not know much about code that I migrate but I see that it intentionally minitors GQL_START and there are some comments about using single WebSocket connection for all subscriptions. The author of this code no longer works for the company. I'm having a hard time figuring out what to do next because I haven't used subscriptions myself. Now I'm thinking of temporarily commenting out the problematic code, migrating the main part, and then returning to the rest when the main part is working. It is possible that everything will be solved easier and I do not see it now.

[Shane32]

Just FYI, in both the old protocol and the new protocol, authentication information should be passed within the initialization message. This was the design and purpose of the initialization message. A major limitation of the prior Server implementation (before my rewrite) was that there was no easy way to access the message payload to authenticate the request. This was doubly-fixed by the two techniques shown above. This was done "twice" because it is common for people to either (a) set and rely on HttpContext.User, such as is needed for the authentication validation rule, and/or (b) hold authentication information within the UserContext.

It is very likely that the prior employee implemented the workaround as you see it because there was no other way to access the initialization payload in that version of GraphQL.NET Server.

[Shane32]

I would suggest implementing one of the techniques that I mentioned, and see if your client applications work properly. If they follow spec, it should not be a problem. I am doubtful of adding interfaces to support out-of-spec clients. In this case the suggested interface has an extremely specific use case. Even the GQL_START message itself (along with the entire subscriptions-transport-ws package) is deprecated. Besides, with creation of a few classes, you can override the methods necessary to support a custom protocol implementation with code injected wherever you need it.

See https://github.com/apollographql/subscriptions-transport-ws

subscriptions-transport-ws was the first implementation of a WebSocket-based GraphQL subscriptions transport in TypeScript. It was created in 2016 and was largely unmaintained after 2018.

Users should migrate to graphql-ws, a newer actively-maintained implementation of a similar protocol.

Note that the original design of Server 7 included an interface solely used to select a subscription server implementation for use. This would allow a user to add (or replace) a subscription protocol without having a custom http middleware implementation. During the design review process, we eliminated this interface, opting for simplicity. You can review this old design here:

https://github.com/Shane32/GraphQL.AspNetCore3/blob/3.0.0/src/GraphQL.AspNetCore3/WebSockets/WebSocketHandler.cs

I think the current design is better, considering the limited use cases for custom subscription server implementations.

[Shane32]

So IWebSocketAuthenticationService is a extension point itself to hook up explicitly into OnConnectionInitAsync - particular single event.

Essentially, yes. But also note that the subscription server also implements proper error handling prior to running that code. Specifically, if GQL_CONNECTION_INIT was sent a second time over the same websocket connection, an error would be sent and the connection terminated, pursuant to protocol.

If the server receives more than one ConnectionInit message at any given time, the server will close the socket with the event 4429: Too many initialisation requests.

[sungam3r]

So far I have commented out the problem code and will return later when I will have repro to experiment with.

[Shane32]

@sungam3r Can we close this? GraphQL.NET Server already has a hook designed to hook into GQL_CONNECTION_INIT based on the spec. And if the user is not designing to spec, it's not that hard to write a custom message processor to handle messages as desired.