BenTools\MercurePHP\Controller\SubscriptionsController

Complexity

Total Complexity

11

Size/Duplication

Total Lines	124
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	filterSubscriptions()	0	20	2
B	__invoke()	0	66	6
A	sendResult()	0	16	1
A	matchRequest()	0	3	1
A	__construct()	0	4	1

<?php

namespace BenTools\MercurePHP\Controller;

use BenTools\MercurePHP\Exception\Http\AccessDeniedHttpException;
use BenTools\MercurePHP\Hub\Hub;
use BenTools\MercurePHP\Model\Subscription;
use BenTools\MercurePHP\Security\Authenticator;
use BenTools\MercurePHP\Security\TopicMatcher;
use BenTools\MercurePHP\Storage\StorageInterface;
use Psr\Http\Message\RequestInterface;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\Http\Message\Response;
use React\Stream\ThroughStream;
use React\Stream\WritableStreamInterface;

use function BenTools\MercurePHP\nullify;

final class SubscriptionsController extends AbstractController
{
    private const PATH = '/.well-known/mercure/subscriptions';
    private const TOPIC_SELECTOR = '/.well-known/mercure/subscriptions/{topic}/{subscriber}';
    private Hub $hub;
    private Authenticator $authenticator;

    public function __construct(Hub $hub, Authenticator $authenticator)
    {
        $this->hub = $hub;
        $this->authenticator = $authenticator;
    }

    public function __invoke(ServerRequestInterface $request): ResponseInterface
    {
        if (!\in_array($request->getMethod(), ['GET', 'HEAD', 'OPTIONS'])) {
            return new Response(405);
        }

        $path = $request->getUri()->getPath();
        $filters = \explode('/', \trim(\strtr($path, [self::PATH => '']), '/'), 2);
        $topic = nullify($filters[0]);
        $subscriber = nullify($filters[1] ?? null);

        $token = $this->authenticator->authenticate($request);

        if (null === $token) {
            throw new AccessDeniedHttpException('You must be authenticated to access the Subscription API.');
        }

        $claim = (array) $token->getClaim('mercure');
        $allowedTopics = $claim['subscribe'] ?? [];
        $deniedTopics = $claim['subscribe_exclude'] ?? [];
        $topicSelector = $path === self::PATH ? self::TOPIC_SELECTOR : $path;
        $matchAllowedTopics = TopicMatcher::matchesTopicSelectors($topicSelector, $allowedTopics);
        $matchDeniedTopics = TopicMatcher::matchesTopicSelectors($topicSelector, $deniedTopics);
        if (!$matchAllowedTopics || $matchDeniedTopics) {
            throw new AccessDeniedHttpException('You are not authorized to display these subscriptions.');
        }

        $stream = new ThroughStream();
        $this->hub->hook(
            function () use ($stream, $path, $subscriber, $topic, $allowedTopics, $deniedTopics) {
                $this->hub->getLastEventID()->then(
                    function (?string $lastEventId) use (
                        $topic,
                        $subscriber,
                        $stream,
                        $path,
                        $allowedTopics,
                        $deniedTopics
                    ) {
                        $this->hub->getActiveSubscriptions($topic, $subscriber)
                            ->then(
                                function (iterable $subscriptions) use (
                                    $stream,
                                    $path,
                                    $allowedTopics,
                                    $deniedTopics,
                                    $lastEventId
                                ) {
                                    $subscriptions = $this->filterSubscriptions(
                                        $subscriptions,
                                        $allowedTopics,
                                        $deniedTopics
                                    );
                                    $this->sendResult($path, $lastEventId, $subscriptions, $stream);
                                }
                            );
                    }
                );
            }
        );

        $headers = [
            'Content-Type' => 'application/ld+json',
        ];

        return new Response(200, $headers, $stream);
    }

    public function matchRequest(RequestInterface $request): bool
    {
        return 0 === \strpos($request->getUri()->getPath(), self::PATH);
    }

    private function filterSubscriptions(iterable $subscriptions, array $allowedTopics, array $deniedTopics): array
    {
        $subscriptions = \iterable_to_array($subscriptions);
        $subscriptions = \array_filter(
            $subscriptions,
            function (Subscription $subscription) use ($allowedTopics, $deniedTopics) {
                $matchAllowedTopics = TopicMatcher::matchesTopicSelectors(
                    $subscription->getId(),
                    $allowedTopics
                );
                $matchDeniedTopics = TopicMatcher::matchesTopicSelectors(
                    $subscription->getId(),
                    $deniedTopics
                );

                return $matchAllowedTopics && !$matchDeniedTopics;
            }
        );

        return \array_values($subscriptions);
    }

    private function sendResult(
        string $path,
        ?string $lastEventId,
        array $subscriptions,
        WritableStreamInterface $stream
    ): void {
        $result = [
            '@context' => 'https://mercure.rocks/',
            'id' => $path,
            'type' => 'Subscriptions',
            'lastEventID' => $lastEventId ?? StorageInterface::EARLIEST,
            'subscriptions' => $subscriptions,
        ];
        $stream->write(\json_encode($result, \JSON_THROW_ON_ERROR));
        $stream->end();
        $stream->close();
    }
}