BenTools\MercurePHP\Controller\SubscribeController

Complexity

Total Complexity

30

Size/Duplication

Total Lines	210
Duplicated Lines	0 %

Test Coverage

Coverage

47.73%

Importance

Changes

0

12 Methods

Rating	Name	Duplication	Size	Complexity
A	dispatchSubscriptions()	0	14	2
A	fetchMissedMessages()	0	7	2
A	createSubscriptions()	0	24	5
A	matchRequest()	0	4	2
A	subscribe()	0	27	4
A	send()	0	8	1
A	__invoke()	0	26	2
A	sendIfAllowed()	0	9	2
A	getLastEventID()	0	6	1
A	withAttributes()	0	34	5
A	sendMissedMessages()	0	12	3
A	__construct()	0	7	1

<?php

namespace BenTools\MercurePHP\Controller;

use BenTools\MercurePHP\Configuration\Configuration;
use BenTools\MercurePHP\Exception\Http\AccessDeniedHttpException;
use BenTools\MercurePHP\Exception\Http\BadRequestHttpException;
use BenTools\MercurePHP\Helpers\QueryStringParser;
use BenTools\MercurePHP\Model\Subscription;
use BenTools\MercurePHP\Security\Authenticator;
use BenTools\MercurePHP\Security\TopicMatcher;
use BenTools\MercurePHP\Model\Message;
use Lcobucci\JWT\Token;
use Psr\Http\Message\RequestInterface;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface as Request;
use Ramsey\Uuid\Uuid;
use React\EventLoop\Factory;
use React\EventLoop\LoopInterface;
use React\Http\Message\Response;
use React\Promise\PromiseInterface;
use React\Stream\ThroughStream;
use React\Stream\WritableStreamInterface as Stream;

use function BenTools\MercurePHP\nullify;
use function BenTools\QueryString\query_string;
use function React\Promise\all;
use function React\Promise\any;
use function React\Promise\resolve;

final class SubscribeController extends AbstractController
{
    private Authenticator $authenticator;
    private LoopInterface $loop;
    private QueryStringParser $queryStringParser;
    private bool $allowAnonymous;

    public function __construct(array $config, Authenticator $authenticator, ?LoopInterface $loop = null)
    {
        $this->config = $config;
        $this->allowAnonymous = $config[Configuration::ALLOW_ANONYMOUS];
        $this->authenticator = $authenticator;
        $this->queryStringParser = new QueryStringParser();
        $this->loop = $loop ?? Factory::create();
    }

    public function __invoke(Request $request): ResponseInterface
    {
        $request = $this->withAttributes($request);

        if ('OPTIONS' === $request->getMethod()) {
            return new Response(200);
        }

        $stream = new ThroughStream();

        $lastEventID = $request->getAttribute('lastEventId');
        $subscribedTopics = $request->getAttribute('subscribedTopics');
        $this->loop
            ->futureTick(
                fn() => $this->dispatchSubscriptions($request->getAttribute('subscriptions'))
                    ->then(fn() => $this->fetchMissedMessages($lastEventID, $subscribedTopics))
                    ->then(fn(iterable $messages) => $this->sendMissedMessages($messages, $request, $stream))
                    ->then(fn() => $this->subscribe($request, $stream))
            );

        $headers = [
            'Content-Type' => 'text/event-stream',
            'Cache-Control' => 'no-cache',
        ];

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

    public function matchRequest(RequestInterface $request): bool
    {
        return \in_array($request->getMethod(), ['GET', 'OPTIONS'], true)
            && '/.well-known/mercure' === $request->getUri()->getPath();
    }

    private function withAttributes(Request $request): Request
    {
        try {
            $token = $this->authenticator->authenticate($request);
        } catch (\RuntimeException $e) {
            throw new AccessDeniedHttpException($e->getMessage());
        }

        if (null === $token && false === $this->allowAnonymous) {
            throw new AccessDeniedHttpException('Anonymous subscriptions are not allowed on this hub.', 401);
        }

        $qs = query_string($request->getUri(), $this->queryStringParser);
        $subscribedTopics = \array_map('\\urldecode', $qs->getParam('topic') ?? []);

        if ([] === $subscribedTopics) {
            throw new BadRequestHttpException('Missing "topic" parameter.');
        }

        $subscriptions = $this->createSubscriptions(
            $subscribedTopics,
            $request->getAttribute('clientId'),
            $token
        );

        $request = $request
            ->withQueryParams($qs->getParams())
            ->withAttribute('token', $token)
            ->withAttribute('subscribedTopics', $subscribedTopics)
            ->withAttribute('lastEventId', $this->getLastEventID($request, $qs->getParams()))
            ->withAttribute('subscriptions', $subscriptions)
        ;

        return $request;
    }

    private function createSubscriptions(array $subscribedTopics, string $clientId, ?Token $token): array
    {
        if (false === $this->config[Configuration::SUBSCRIPTIONS]) {
            return [];
        }

        $subscriptions = [];
        foreach ($subscribedTopics as $subscribedTopic) {
            $id = \sprintf('/.well-known/mercure/subscriptions/%s/%s', \urlencode($subscribedTopic), $clientId);
            try {
                $payload = null !== $token ? ($token->getClaim('mercure')['payload'] ?? null) : null;
            } catch (\Exception $e) {
                $payload = null;
            }
            $subscriptions[] = new Subscription(
                $id,
                $clientId,
                $subscribedTopic,
                true,
                $payload,
            );
        }

        return $subscriptions;
    }

    /**
     * @param Subscription[] $subscriptions
     */
    private function dispatchSubscriptions(array $subscriptions): PromiseInterface
    {
        $promises = [$this->storage->storeSubscriptions($subscriptions)];

The method storeSubscriptions() does not exist on null.

        foreach ($subscriptions as $subscription) {
            $message = new Message(
                (string) Uuid::uuid4(),
                \json_encode($subscription, \JSON_THROW_ON_ERROR),
                true,
            );
            $topic = $subscription->getId();
            $promises[] = $this->transport->publish($topic, $message);

The method publish() does not exist on null.

        }

        return any($promises);
    }

    private function subscribe(Request $request, Stream $stream): PromiseInterface
    {
        $subscribedTopics = $request->getAttribute('subscribedTopics');
        $token = $request->getAttribute('token');
        $promises = [];
        foreach ($subscribedTopics as $topicSelector) {
            if (!TopicMatcher::canSubscribeToTopic($topicSelector, $token, $this->allowAnonymous)) {
                $clientId = $request->getAttribute('clientId');
                $this->logger()->debug("Client {$clientId} cannot subscribe to {$topicSelector}");
                continue;
            }
            $promises[] = $this->transport
                ->subscribe(
                    $topicSelector,
                    fn(string $topic, Message $message) => $this->sendIfAllowed($topic, $message, $request, $stream)
                )
                ->then(function (string $topic) use ($request) {
                    $clientId = $request->getAttribute('clientId');
                    $this->logger()->debug("Client {$clientId} subscribed to {$topic}");
                });
        }

        if ([] === $promises) {
            return resolve(true);
        }

        return all($promises);
    }

    private function fetchMissedMessages(?string $lastEventID, array $subscribedTopics): PromiseInterface
    {
        if (null === $lastEventID) {
            return resolve([]);
        }

        return $this->storage->retrieveMessagesAfterId($lastEventID, $subscribedTopics);
    }

    private function sendMissedMessages(iterable $messages, Request $request, Stream $stream): PromiseInterface
    {
        $promises = [];
        foreach ($messages as $topic => $message) {
            $promises[] = $this->sendIfAllowed($topic, $message, $request, $stream);
        }

        if ([] === $promises) {
            return resolve(true);
        }

        return all($promises);
    }

    private function sendIfAllowed(string $topic, Message $message, Request $request, Stream $stream): PromiseInterface
    {
        $subscribedTopics = $request->getAttribute('subscribedTopics');
        $token = $request->getAttribute('token');
        if (!TopicMatcher::canReceiveUpdate($topic, $message, $subscribedTopics, $token, $this->allowAnonymous)) {
            return resolve(false);
        }

        return resolve($this->send($topic, $message, $request, $stream));
    }

    private function send(string $topic, Message $message, Request $request, Stream $stream): PromiseInterface
    {
        $stream->write((string) $message);
        $clientId = $request->getAttribute('clientId');
        $id = $message->getId();
        $this->logger()->debug("Dispatched message {$id} to client {$clientId} on topic {$topic}");

        return resolve(true);
    }

    private function getLastEventID(Request $request, array $queryParams): ?string
    {
        return nullify($request->getHeaderLine('Last-Event-ID'))
            ?? nullify($queryParams['Last-Event-ID'] ?? null)
            ?? nullify($queryParams['Last-Event-Id'] ?? null)
            ?? nullify($queryParams['last-event-id'] ?? null);
    }
}