Exposing custom GQL Subscriptions

Subscriptions let clients open a long-lived connection and receive an event every time something interesting happens on the server, instead of having to poll a query. They're exposed the same way Queries and Mutations are: by declaring a class, very similar to an App Service, that implements a marker interface — ISubscription in this case — which the GraphQL module auto-discovers and adds to the schema.

public class FoldersChangedSubscription : ISubscription
{
    [Subscribe]
    [Topic("foldersChanged")]
    public FoldersChanged FoldersChanged([EventMessage] FoldersChanged folderFoldersChanged)
        => folderFoldersChanged;
}

The [Subscribe] attribute marks the method as the subscription's resolver, and [Topic] tells HotChocolate which topic to listen on. Whatever gets sent to that topic (see Publishing events below) is injected into the method through [EventMessage] and returned to the client as-is.

Note

Just like Queries and Mutations, a subscription class is only exposed on a given service's schema if the assembly it lives in is part of that service's module dependency graph — each service authors and discovers its own subscription resolvers, there's no sharing of resolver code across projects. That subgraph schema is still composed into the gateway like any other, though: see Subscriptions and Fusion below.

Templated topics

Topics can be parameterized from the method's arguments using HotChocolate's {argName} syntax. This is how LiveUpdateSubscriptions lets a client subscribe to updates for a specific entity or channel without us having to declare a fixed topic per id:

public class LiveUpdateSubscriptions : ISubscription
{
    [Subscribe]
    [Topic($"{LiveUpdateWellKnownConstants.LiveUpdatePrefix}{{id}}")]
    public LiveUpdateEvent LiveUpdateEventsById(
        string id,
        [EventMessage] LiveUpdateEvent liveUpdateMessage)
    {
        return liveUpdateMessage;
    }

    [Subscribe]
    [Topic($"{LiveUpdateWellKnownConstants.LiveUpdatePrefix}{{channel}}")]
    public LiveUpdateEvent LiveUpdateEvents(
        string channel,
        [EventMessage] LiveUpdateEvent liveUpdateMessage)
    {
        return liveUpdateMessage;
    }
}

Subscribing to multiple topics

Sometimes the topic to listen on can't be derived from a single templated argument — for example, a notifications subscription that needs to listen for a given user across several app tags at once. For these cases we write a custom SubscribeAsync method instead of relying on [Topic], and use ITopicEventReceiver directly to open the underlying stream:

public class CreateNotificationSubscription : ISubscription
{
    [Subscribe(With = nameof(SubscribeAsync))]
    public Notification? NotificationsForUser(
        Guid userId,
        string[]? appTags,
        [EventMessage] Notification notification,
        IResolverContext context)
    {
        context.SetCurrentUserFromContextOrDefault(userId);
        return notification;
    }

    public async ValueTask<ISourceStream<Notification>> SubscribeAsync(
        Guid userId,
        string[]? appTags,
        [Service] ITopicEventReceiver receiver,
        CancellationToken cancellationToken)
    {
        var topics = SubscriptionTopics.GetNotificationReceivedSubscribeTopics(userId, appTags);
        return await receiver.SubscribeToMultipleTopicsAsync<Notification>(topics, cancellationToken);
    }
}

SubscribeToMultipleTopicsAsync is a helper provided by the GraphQL module that subscribes to every topic in the list and merges them into a single event stream, so the resolver method doesn't need to know how many topics it ended up listening to.

Publishing events

Subscriptions only describe how clients listen; something on the server still has to push an event onto a topic for a subscriber to receive anything. This is done with ITopicEventSender.SendAsync, and in practice it's wired up from a MassTransit consumer, since that's usually where we learn that something worth notifying about just happened.

The simplest form is a dedicated consumer that reacts to a message and sends straight to the matching topic:

public class ClaimCostUpdatedConsumer(
    ITopicEventSender topicEventSender, ISuiteContext suiteContext, ILogger<ClaimCostUpdatedConsumer> logger)
    : IConsumer<ClaimCostUpdated>
{
    public async Task Consume(ConsumeContext<ClaimCostUpdated> context)
    {
        // [..]
        var topic = SubscriptionTopics.GetCostSummaryTopic(data.CorrelationId, currentUserId.Value);
        await topicEventSender.SendAsync(topic, payload);
    }
}

The publisher and the subscriber are only coupled by agreeing on the same topic string. There's no compile-time link between them, so keeping topic construction in a shared helper (like SubscriptionTopics) is important to avoid the two sides drifting apart.

Using Live Updates

For services that just want to notify subscribers "this entity changed" for a set of existing messages, without writing a bespoke consumer per message type, Suite provides LiveUpdatesClientModule — a generic framework that intercepts MassTransit publishes, extracts the entity id/channel, and broadcasts it as a subscription event.

Setup is configuration-only. A service just depends on the module and declares which messages to notify on:

builder.DependsOn<LiveUpdatesClientModule, LiveUpdatesClientModuleOptions>(options =>
{
    options.NotifyOnMessage<OperationCompleted, Operation>();
    options.NotifyOnMessage<OperationFailed, Operation>();
    options.NotifyOnMessage<OperationStarted, Operation>();
});

Under the hood, a MassTransit publish filter (LiveUpdateMessageFilter<TMessage>) intercepts each outbound message, extracts the entity id and event type, wraps it in a LiveUpdateEvent, and calls ITopicEventSender through the SubscriptionLiveUpdateSender abstraction:

public class SubscriptionLiveUpdateSender(ITopicEventSender topicEventSender, ILogger<SubscriptionLiveUpdateSender> logger)
    : ILiveUpdateSender
{
    public async Task SendToTargetAsync(LiveUpdateEvent liveUpdateEvent) =>
        await topicEventSender.SendAsync(
            $"{LiveUpdateWellKnownConstants.LiveUpdatePrefix}{liveUpdateEvent.TargetId}", liveUpdateEvent);

    public async Task SendToChannelAsync(LiveUpdateEvent liveUpdateEvent) =>
        await topicEventSender.SendAsync(
            $"{LiveUpdateWellKnownConstants.LiveUpdatePrefix}{liveUpdateEvent.Channel}", liveUpdateEvent);
}

On the subscribe side, LiveUpdateSubscriptions declares two fields — liveUpdateEventsById(id) and liveUpdateEvents(channel) — that listen to matching topics. Because LiveUpdatesClientModule already depends on GraphQLModule with EnableSubscriptions = true, this single dependency is enough to get both sides: the publish filter and the subscription resolvers.

Backplane

Subscriptions need a pub/sub backplane so that an event sent from one server instance reaches a client connected to a different instance. This is controlled by GraphQLModuleOptions.SubscriptionsProviderType, which defaults to Redis:

options.SubscriptionsProviderType = SubscriptionsProviderType.Redis;

InMemory is also available, but only delivers events within a single process, so it's mainly useful for local development or testing.

Subscriptions and Fusion

Subscriptions are composed into a gateway's schema exactly the same way Queries and Mutations are — see Fusion for the general composition model. A subgraph's schema.graphql simply declares a Subscription root type alongside Query and Mutation, and Fusion merges it into the gateway using the same field-name-based composition it uses everywhere else:

type Subscription {
  badgeCountUpdated(appTags: [String!] userId: UUID!): BadgeCountUpdated!
    @variable(subgraph: "notifications", name: "userId", argument: "userId")
    @resolver(
      subgraph: "notifications"
      select: "{ badgeCountUpdated(userId: $userId, appTags: $appTags) }"
      kind: "SUBSCRIBE"
    )
}

The kind: "SUBSCRIBE" resolver directive is what tells the gateway this field needs a persistent connection to the subgraph instead of a one-shot request — everything else about the directive (@variable, argument forwarding) works the same as a Query or Mutation resolver.

Because composition merges by field name, the same subscription field can even be backed by more than one subgraph at once. foldersChanged is declared independently by three different services — LandingRequests, UxRoles and Tracking — each with their own [Subscribe]/[Topic] resolver class, and Fusion collapses them into a single gateway field with one resolver per source:

foldersChanged: FoldersChanged!
  @resolver(subgraph: "landing_requests", select: "{ foldersChanged }", kind: "SUBSCRIBE")
  @resolver(subgraph: "ux_roles", select: "{ foldersChanged }", kind: "SUBSCRIBE")
  @resolver(subgraph: "tracking", select: "{ foldersChanged }", kind: "SUBSCRIBE")

A client subscribing to foldersChanged on the gateway transparently receives events from whichever of the three subgraphs fires first — it has no idea, and doesn't need to know, that three independent services are behind that one field.

Wiring a subgraph's subscriptions for composition

For the gateway to actually open that persistent connection, each subgraph's subgraph-config.json needs a websocket address in addition to the regular http one:

{
    "subgraph": "notifications",
    "http": { "baseAddress": "http://notifications-service/graphql" },
    "websocket": { "baseAddress": "ws://notifications-service/graphql" }
}

On the gateway side, GraphQLGatewayModule depends on WebSocketModule and RedisModule unconditionally, so any gateway is always ready to both accept client subscriptions over its own WebSocket endpoint and open outbound WebSocket connections to whichever subgraphs it needs to reach.