Integration of subscriptions

[WizMik]

This is F*CKING awesome !
I wanted to tell you that while writing the first issue, which is more a feature request.
This package only misses the integration of subscriptions requests to be perfect.

Meantime I still have a lot of new toys to play with, so thanks!

[moufmouf]

Subscriptions could actually be added!

My current plan is to try the Mercure protocol for real-time communication.

[oprypkhantc]

I'm interested in this as well. WebSockets require a separate web server, separate configuration for proxying servers (like Nginx).

What about SSE? Stream responses are already supported by Laravel, Symfony, Swoole and are as easy to use on the client as WebSockets are. Their performance under stress conditions is lower than that of WebSocket, but the conclusion of TimePlus' article is that they're "close enough".

This might be a less painful option for graphqlite's users than WS, thanks to everything being in one app, on one server and in PHP.

[oojacoboo]

@oprypkhantc it'd be nice to see some action around this for sure. I agree that sticking with SSE and the HTTP protocol is more fitting for GraphQLite. Bi-directional communication over WS isn't really necessary for GraphQL.

API Platform has implemented Mercure support for subscriptions
api-platform/core#3321

Here is a small SSE proof of concept app using Mercure and symfony/mercure
https://github.com/AlessandroMinoccheri/sse-poc

Here is a topic with webonyx on subscription implementation
webonyx/graphql-php#9

Webonyx already supports the subscription property on the GraphQL\Type\Schema::__construct(array $config). So, I'm assuming we could just introduce a new GraphQLite\Subscription attribute to annotate the "controllers"/resolvers. We'd likely want to have some abstractions around symfony/mercure, for publishing updates, as that'll require building out the types/fields for the payload, based on the initial client's subscription query.

What I'm unsure about is, when a subscription is made, presumably with an accompanying query (output fields needed - don't see how input fields could be allowed) - where is that query stored for later use? I'm assuming that'd require a custom userland persistence implementation, which is fine.

Maybe as a first step we could outline the API and clear up some of the unknowns with how an implementation might be done.

[oprypkhantc]

I believe the initial effort should be focused on providing a stupid-simple contract that users could then use to build more sophisticated solutions. For example, ExpediaGroup/graphql-kotlin uses existing Java ecosystem to handle subscriptions on the most basic level: https://github.com/ExpediaGroup/graphql-kotlin/blob/master/examples/server/spring-server/src/main/kotlin/com/expediagroup/graphql/examples/server/spring/subscriptions/SimpleSubscription.kt#L39

The way it works is they have two interfaces: Publisher and Subscriber:

interface Subscriber<TItem> {
    public function onSubscribe​(Subscription $subscription): void;
    public function onNext​(TItem $item): void;
    public function onError(Throwable $error): void;
    public function onComplete(): void;
}

interface Publisher<TItem> {
    public function subscribe(Subscriber<TItem> $subscriber): void;
}

And this is how an example implementation might look:

class EverySecondPublisher implements Publisher<int> {
    public function subscribe(Subscriber<int> $subscriber): void
    {
        $subscriber->onSubscribe();
        
        for ($i = 0; $i < 1000; $i++) {
            $subscriber->onNext($i);
        }
        
        $subscriber->onComplete();
    }
}

class SomeController {
    /** @return Publisher<int> */
    #[Subscription]
    public function intAddedEverySecond(): EverySecondPublisher {
        return new EverySecondPublisher();
    }
}

Then all that's left is to implement Subscriber for SSE and use it:

// Graphqlite internals...

class SSESubscriptionFromPsrRequest implements Subscriber {
    public function __construct(private readonly RequestInterface $request) {}
    
    public function onSubscribe(): void {
        // send HTTP headers ?
        $request->open();
    }
    
    public function onNext($item): void {
        $serialized = json_encode($item);
        
        $request->send($serialized);
    }
    
    public function onComplete(): void {
        $request->close();
    }
}

$publisher = $controller->callSubscriptionMethod();
$publisher->subscribe(new SSESubscriptionFromPsrRequest($request));

then, on top of it, we could implement first-party support for symfony/mercure on top of those interfaces.

But I advocate against it. That can be easily included in the graphqlite/symfony-bundle if needed. I'd rather leave that to be framework-specific; for example, in Laravel the standard practice is to use broadcastable events:

class MaintenanceModeEvent {
    public function broadcastChannel(): string {
        return 'maintenance';
    }
}

// Broadcast to every subscriber of `general` channel
$broadcaster->broadcast(new MaintenanceModeEvent());

So instead of changing events throughout the project or trying to workaround symfony/mercure, I'd much rather just implement an adapter to use the existing Laravel's system:

class LaravelBroadcasterPublisher implements Publisher {
    public function __construct(private readonly string $channel, private readonly Broadcaster $broadcaster) {}
    
    public function subscribe(Subscriber $subscriber): void 
    {
        $subscriber->onSubscribe();
        
        // example code, not a real API
        $broadcaster->listen($this->channel, function ($event) use ($subscriber) {
            $subscriber->onNext($event);
        });
    }
}

class SomeController {
    #[Subscription]
    public function maintenanceMode(#[Inject] Broadcaster $broadcaster): Publisher {
        return new LaravelBroadcasterPublisher('maintenance', $broadcaster);
    }
}

Of course this will also have to account for async use cases (through the use of graphql-php promises I guess),

[oojacoboo]

@oprypkhantc Something I'm unsure about is the return type of the subscription resolver. Currently we're using this for the output type and fields. We might be able to override this by setting the outputType on the Subscription attribute, but I don't love that. In this example, what does returning the Publisher even do, if we're to assume that it's opening a stream? Also, have to think about the input types, which would also need to be used for building the schema definition for the subscription.

Another way we could look at this is that each subscription resolver would look like our query and mutation resolvers, with input type arguments and a proper output types. Then, this method/operation/subscription would be executed when broadcasting. The actual subscription/pub-sub logic is handled as a separate service. Currently, as it is now, I don't see how we're going to define the definition schema for the subscription. And it's way more likely that the pub-sub logic is going to be uniform.

[oprypkhantc]

@oprypkhantc Something I'm unsure about is the return type of the subscription resolver. Currently we're using this for the output type and fields. We might be able to override this by setting the outputType on the Subscription attribute, but I don't love that. In this example, what does returning the Publisher even do, if we're to assume that it's opening a stream? Also, have to think about the input types, which would also need to be used for building the schema definition for the subscription.

Output types are there, they're just nested as a generic type of Publisher. In one of the examples the return type is specified as /** @return Publisher<int> */, and that's how GraphQLite will be able to map the output type automatically. As for the input types, they should simply work the same way they do for regular queries and mutations:

class SomeController {
    /** @return Publisher<String> */
    #[Subscription]
    public function stringEveryNSeconds(int $numberOfSeconds): Publisher {}
}

type Subscription {
    stringEveryNSeconds(numberOfSeconds: Int): String
}

Regarding the Publisher type, it's not immediately returning a stream. It's returning a wrapper that can open a stream whenever you (graphqlite) are ready to open it, which might depend on the implementation, by calling the subscribe method. Think of it as a "OpenableStream". We cannot immediately return a stream because we need access to the Subscriber object, and exposing it through the Publisher return type is the only type-safe way of doing so. It's a good design to steal from the Java ecosystem 😆

And to clarify, this API is more of something to be built upon rather than something to be used by end users. It allows to build anything we want on top of it, but something of this kind should be the foundation of subscriptions so users can implement their own annotations, middleware and whatnot to interact with subscriptions the way they want.

The actual subscription/pub-sub logic is handled as a separate service.

That we can do on top of the publisher/subscriber easily:

class SomeController {
    /** @return Publisher<Post> */
    #[Subscription]
    public function postUpdated(#[UseInputType('ID!')] string $post): Publisher {
        return new LaravelBroadcasterPublisher(
            PostUpdatedEvent::class, 
            // filter for posts by input arguments
            fn (PostUpdatedEvent $event) => $event->post->id === $post
        );
    }
}

If you then want something even more abstract, it could be done with middleware and annotations:

class SomeController {
    #[Subscription]
    #[SubscribeToBroadcaster(PostUpdatedEvent::class)]
    public function postUpdated(#[UseInputType('ID!')] string $post): ?callable {
        return fn (PostUpdatedEvent $event) => $event->post->id === $post;
    }
}

class SubscribeToBroadcaster {
    public function __construct(public readonly string $event) {}
}

class SubscribeToBroadcasterMiddleware {
    public function process($fieldDescriptor, $fieldHandler): FieldDefinition|null
    {
        $subscribeToBroadcasterAttribute = $fieldDescriptor->getMiddlewareAnnotations()->getAnnotationByType(SubscribeToBroadcaster::class);

        if (! $subscribeToBroadcasterAttribute) {
            return $fieldHandler->handle($fieldDescriptor);
        }
        
        // yada yada, map return type from PostUpdatedEvent::class
        $returnType = mapOutputType(reflectionReturnType($subscribeToBroadcasterAttribute->event, 'broadcastAs'));
        
        $fieldDescriptor = $fieldDescriptor
            ->withType($returnType)
            ->withResolver(function () {
                $filter = ($fieldDescriptor->getResolver())();
                
                return new LaravelBroadcasterPublisher($subscribeToBroadcasterAttribute->event, $filter);
            });
            
        return $fieldHandler->handle($fieldDescriptor);
    }
}

[oojacoboo]

This really feels like it's being attacked from the wrong angle - almost like trying to shoehorn something into an existing design.

An annotation for the return type is certainly undesirable. I realize we have to do that with iterables currently, but that's more of a PHP limitation than a design decision.

What is the benefit of trying to make this subscription so coupled with this pub/sub implementation?

Taking a more pure approach to this, where does this limit implementation with a pub/sub design like you're proposing?

class Controller
{
    
    #[Subscription(
        subscriber: Subscriber::class, // Optional - can have a main config global subscription handler service registered
    )]
    public function articlePublished(ArticleFilterInput $filter): ?Article
    {
        // Some code that returns the published article based on the filters
    }
}

The subscriber argument of the Subscription attribute would implement a subscriber interface, the same one registered globally that would apply to all controller/resolver methods annotated with the Subscription attribute.

And that's basically it. The Subscription attribute and associated method would be used for definition schema building. The subscriber would be responsible for managing the subscription entirely. The whole pub/sub behavior can be contained there. The controller/resolver method just serves to be a pure representation of each execution/broadcast.

[oprypkhantc]

@oojacoboo What we do with iterables is a PHP limitation, but so is this. Generics are part of the language now, even with the ugly PHPDoc syntax.

I'm not sure I'm following. How do I replicate my example that emits a counter every second (without global state, global broadcaster or other services running)?

Also articlePublished return type is specified as ?Article, but how do I return multiple "events" (i.e. multiple broadcasts of the new Article)? Also, how would this code wait for the new article to arrive?

Let's say we have this code:

class ArticleController {
    #[Mutation]
    public function createArticle(ArticleInput $data): Article {
        $article = Article::fromInput($data);
        
        $this->save($article);

        $this->events->dispatch(new ArticleCreatedEvent($article)); // let's say it's also published into Redis

        return $article;
    }

    #[Subscription()]
    public function articlePublished(ArticlePublishedInput $filters): ?Article
    {
        // how do I subscribe to ArticleCreatedEvent here, given that I can subscribe to Redis channel and get  an ArticleCreatedEvent instance with article inside of it?
        // how do I return all subsequently created articles, not just the first one?
    }
}

[oojacoboo]

So, there would be a global Subscriber service, which would implement our SubscriberInterface, that'd be registered with the GraphQLite config. Also, the Subscription attribute would take an optional subscriber argument to specify an optional custom subscriber per subscription.

class CustomSubscriber implements SubscriberInterface {}

class ArticleController {
    #[Subscription(
        subscriber: CustomSubscriber::class, // Guess this could be an object as well implementing the SubscriberInterface
    )]
    public function articlePublished(ArticlePublishedInput $filters): ?Article
    {
        // how do I subscribe to ArticleCreatedEvent here, given that I can subscribe to Redis channel and get  an ArticleCreatedEvent instance with article inside of it?
        // how do I return all subsequently created articles, not just the first one?
    }
}

how do I subscribe to ArticleCreatedEvent here, given that I can subscribe to Redis channel and get an ArticleCreatedEvent instance with article inside of it?

You can handle this in your Subscriber, if this is the design you prefer.

how do I return all subsequently created articles, not just the first one?

Your Subscriber would determine whatever is/how you are calling your Subscription annotated controller method each time the event is fired. You can, of course, also return an iterable from your subscription if needed, just like any other operation.

So, basically the flow is this...

Whenever a subscription request is made, GraphQLite recognizes the operation as a subscription, validates the request/GQL against the schema, and checks the Subscription attribute associated with the controller method matching the field (articlePublished), to see if there is a custom Subscriber defined (in this case there is). If so, it'll use that custom Subscriber (CustomSubscriber), else it'll use the default Subscriber defined in the global GraphQLite config.

At this point, your Subscriber can do whatever it likes basically. In your case, you're just pushing your global state to Redis. Someone else might push to a database, or if they're using a long running server, maybe even keep it in memory. Then, whenever an event is fired in your system, it'd just execute the controller method and return/push the result to the appropriate client/channel/steam/endpoint/etc.

As for what the Subscriber would receive as a payload - I assume it'd need the AST/GQL and variables. What I'm unsure about is subscription operations with multiple top-level fields, like:

subscription ArticlesAdded ($type: String!) {
    articlePublished(filter: {
        type: $type
    }) {
        ...fields
    }

    articlePublished(filter: {
        type: $type
    }) {
        ...fields
    }
}

We could limit subscription operations to a single field. That'd then allow us to add the articlePublished value to the payload reliably, if that's event useful. The payload could also potentially include a callable, or some other serialized object. I think the raw GQL is going to be the best though.

[oprypkhantc]

So, there would be a global Subscriber service, which would implement our SubscriberInterface, that'd be registered with the GraphQLite config. Also, the Subscription attribute would take an optional subscriber argument to specify an optional custom subscriber per subscription.

I'm still not getting it though. Can you implement the SubscriberInterface interface and controller methods for both examples (a ticking counter and article publish)?

You can handle this in your Subscriber, if this is the design you prefer.
Your Subscriber would determine whatever is/how you are calling your Subscription annotated controller method each time the event is fired. You can, of course, also return an iterable from your subscription if needed, just like any other operation.

I don't quite understand. If I have to pull the article from Redis in my subscriber class, then why do I need the controller at all and when/why do I call it?

class PublishedArticleRedisSubscriber implements Subscriber {
    public function subscribe(ArticleFilterInput $input, GraphQLiteSubscription $subscription): {
        // I need deserialized input here to be able to subscribe to the correct channel
        $this->redis->sub(`article.published.{$input->type}`, function ($payload) {
            $event = deserialize($payload);
        
            // When and where do I need to call the controller?
            // Also take into account that this might get called a thousand times 
            // before the subscription's over, so calling the controller method with 
            // it's return type specified as ?Article isn't an option
            $subscription->onNext($event->article);
        });
    }
}