Merge pull request #144 from buggregator/feauture/124

Adds webhook module

.rr-prod.yaml

@@ -47,6 +47,9 @@ kv:
     driver: memory
     config: {}
 
+jobs:
+  consume: []
+
 service:
   nginx:
     service_name_in_log: true

.rr.yaml

@@ -45,6 +45,9 @@ kv:
     driver: memory
     config: {}
 
+jobs:
+  consume: []
+
 service:
 #  frontend:
 #    service_name_in_log: true

app/config/cache.php

@@ -10,6 +10,7 @@
     'default' => env('CACHE_STORAGE', 'roadrunner'),
     'aliases' => [
         'events' => ['storage' => $defaultStorage, 'prefix' => 'events:'],
+        'webhooks' => ['storage' => $defaultStorage, 'prefix' => 'webhooks:'],
         'local' => ['storage' => $defaultStorage, 'prefix' => 'local:'],
     ],
     'storages' => [

app/config/events.php

@@ -4,6 +4,7 @@
 
 return [
     'interceptors' => [
-        \App\Application\Broadcasting\BroadcastEventInterceptor::class
+        \Modules\Webhooks\Application\Broadcasting\WebhookEventInterceptor::class,
+        \App\Application\Broadcasting\BroadcastEventInterceptor::class,
     ]
 ];

app/config/queue.php

@@ -2,21 +2,22 @@
 
 declare(strict_types=1);
 
+use Modules\Webhooks\Interfaces\Job\WebhookHandler;
 use Spiral\Queue\Driver\SyncDriver;
 use Spiral\RoadRunner\Jobs\Queue\MemoryCreateInfo;
-use Spiral\RoadRunner\Jobs\Queue\AMQPCreateInfo;
-use Spiral\RoadRunner\Jobs\Queue\BeanstalkCreateInfo;
-use Spiral\RoadRunner\Jobs\Queue\SQSCreateInfo;
 use Spiral\RoadRunnerBridge\Queue\Queue;
 
 return [
     'default' => env('QUEUE_CONNECTION', 'sync'),
-    'aliases' => [],
+    'aliases' => [
+        'webhook' => 'roadrunner',
+        'events' => 'roadrunner',
+    ],
     'pipelines' => [
         'memory' => [
             'connector' => new MemoryCreateInfo('local'),
             'consume' => true,
-        ]
+        ],
     ],
     'connections' => [
         'sync' => [
@@ -33,5 +34,8 @@
     ],
     'registry' => [
         'handlers' => [],
+        'serializers' => [
+            WebhookHandler::class => 'symfony-json',
+        ],
     ],
 ];

app/config/swagger.php

@@ -64,5 +64,6 @@
         directory('app') . 'src/Application/HTTP/Response',
         directory('app') . 'src/Interfaces/Http',
         directory('app') . 'modules/Events/Interfaces/Http',
+        directory('app') . 'modules/Webhooks/Interfaces/Http',
     ],
 ];

app/modules/Sentry/Interfaces/Http/Handler/EventHandler.php

@@ -38,8 +38,6 @@ public function handle(ServerRequestInterface $request, \Closure $next): Respons
 
         $payloads = $this->payloadParser->parse($request);
 
-        dump($payloads);
-
         match (true) {
             \str_ends_with($url, '/envelope') => $this->handleEnvelope($payloads),
             \str_ends_with($url, '/store') => $this->handleEvent($payloads),

app/modules/Webhooks/Application/Broadcasting/WebhookEventInterceptor.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application\Broadcasting;
+
+use Modules\Events\Domain\Events\EventWasReceived;
+use Modules\Webhooks\Domain\WebhookEvent;
+use Modules\Webhooks\Domain\WebhookServiceInterface;
+use Spiral\Core\CoreInterceptorInterface;
+use Spiral\Core\CoreInterface;
+
+final readonly class WebhookEventInterceptor implements CoreInterceptorInterface
+{
+    public function __construct(
+        private WebhookServiceInterface $service,
+    ) {
+    }
+
+    public function process(string $controller, string $action, array $parameters, CoreInterface $core): mixed
+    {
+        $event = $parameters['event'];
+
+        $result = $core->callAction($controller, $action, $parameters);
+
+        $webhookEvent = null;
+        if ($event instanceof EventWasReceived) {
+            $webhookEvent = new WebhookEvent(
+                event: $event->type . '.received',
+                payload: [
+                    'uuid' => (string)$event->uuid,
+                    'type' => $event->type,
+                    'payload' => $event->payload,
+                    'timestamp' => $event->timestamp,
+                ],
+            );
+        }
+
+
+        if ($webhookEvent) {
+            $this->service->send($webhookEvent);
+        }
+
+        return $result;
+    }
+}

app/modules/Webhooks/Application/CompositeWebhookLocator.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application;
+
+use Modules\Webhooks\Domain\WebhookLocatorInterface;
+
+final readonly class CompositeWebhookLocator implements WebhookLocatorInterface
+{
+    /**
+     * @param WebhookLocatorInterface[] $locators
+     */
+    public function __construct(
+        private array $locators,
+    ) {
+    }
+
+    public function findAll(): iterable
+    {
+        foreach ($this->locators as $locator) {
+            yield from $locator->findAll();
+        }
+    }
+}

app/modules/Webhooks/Application/DeliveryFactory.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application;
+
+use App\Application\Domain\ValueObjects\Uuid;
+use Modules\Webhooks\Domain\Delivery;
+use Modules\Webhooks\Domain\DeliveryFactoryInterface;
+
+final readonly class DeliveryFactory implements DeliveryFactoryInterface
+{
+    public function create(
+        Uuid $webhookUuid,
+        string $payload,
+        string $response,
+        int $status,
+    ): Delivery {
+        return new Delivery(
+            uuid: Uuid::generate(),
+            webhookUuid: $webhookUuid,
+            payload: $payload,
+            response: $response,
+            status: $status,
+        );
+    }
+}

app/modules/Webhooks/Application/InMemoryDeliveryRepository.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application;
+
+use App\Application\Domain\ValueObjects\Uuid;
+use Modules\Webhooks\Domain\Delivery;
+use Modules\Webhooks\Domain\DeliveryRepositoryInterface;
+use Psr\SimpleCache\CacheInterface;
+use Spiral\Core\Attribute\Singleton;
+
+#[Singleton]
+final readonly class InMemoryDeliveryRepository implements DeliveryRepositoryInterface
+{
+    public function __construct(
+        private CacheInterface $cache,
+        private int $maxDeliveries = 10,
+        private int $ttl = 3600, // 1 hour in seconds
+    ) {
+    }
+
+    public function findAll(Uuid $webhookUuid): array
+    {
+        return $this->cache->get($this->getKey($webhookUuid), []);
+    }
+
+    public function store(Delivery $delivery): void
+    {
+        $deliveries = \array_slice(
+            $this->cache->get($this->getKey($delivery->webhookUuid), []),
+            0,
+            $this->maxDeliveries,
+        );
+
+        $deliveries[] = $delivery;
+
+        $this->cache->set($this->getKey($delivery->webhookUuid), $deliveries, $this->ttl);
+    }
+
+    /**
+     * @return non-empty-string
+     */
+    public function getKey(Uuid $webhookUuid): string
+    {
+        return 'deliveries:' . $webhookUuid;
+    }
+}

app/modules/Webhooks/Application/InMemoryWebhookRepository.php

@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application;
+
+use App\Application\Domain\ValueObjects\Uuid;
+use Modules\Webhooks\Domain\Webhook;
+use Modules\Webhooks\Domain\WebhookRegistryInterface;
+use Modules\Webhooks\Domain\WebhookRepositoryInterface;
+use Modules\Webhooks\Exceptions\WebhookNotFoundException;
+use Psr\Log\LoggerInterface;
+use Psr\SimpleCache\CacheInterface;
+use Spiral\Core\Attribute\Singleton;
+
+#[Singleton]
+final readonly class InMemoryWebhookRepository implements WebhookRepositoryInterface, WebhookRegistryInterface
+{
+    public function __construct(
+        private LoggerInterface $logger,
+        private CacheInterface $cache,
+    ) {
+    }
+
+    public function findByEvent(string $event): array
+    {
+        return \array_filter($this->getWebhooks(), fn(Webhook $webhook) => $webhook->event === $event);
+    }
+
+    public function getByUuid(Uuid $uuid): Webhook
+    {
+        return $this->findByUuid($uuid) ?? throw new WebhookNotFoundException(
+            \sprintf('Webhook with UUID %s not found', $uuid),
+        );
+    }
+
+    public function findByUuid(Uuid $uuid): ?Webhook
+    {
+        return $this->getWebhooks()[(string)$uuid] ?? null;
+    }
+
+    public function register(Webhook $webhook): void
+    {
+        if ($this->findByUuid($webhook->uuid) !== null) {
+            return;
+        }
+
+        $webhooks = $this->getWebhooks();
+        $webhooks[(string)$webhook->uuid] = $webhook;
+
+        $this->logger->debug('Webhook registered', [
+            'uuid' => (string)$webhook->uuid,
+            'event' => $webhook->event,
+            'url' => $webhook->url,
+        ]);
+
+        $this->cache->set('webhooks', $webhooks);
+    }
+
+    /**
+     * @return Webhook[]
+     */
+    private function getWebhooks(): array
+    {
+        return $this->cache->get('webhooks', []);
+    }
+
+    public function findAll(): array
+    {
+        return \array_values($this->getWebhooks());
+    }
+}

app/modules/Webhooks/Application/WebhookFactory.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application;
+
+use App\Application\Domain\ValueObjects\Uuid;
+use Modules\Webhooks\Domain\Webhook;
+use Modules\Webhooks\Domain\WebhookFactoryInterface;
+
+final readonly class WebhookFactory implements WebhookFactoryInterface
+{
+    public function create(
+        string $event,
+        string $url,
+        bool $verifySsl = false,
+        bool $retryOnFailure = true,
+    ): Webhook {
+        return new Webhook(
+            uuid: Uuid::generate(),
+            event: $event,
+            url: $url,
+            verifySsl: $verifySsl,
+            retryOnFailure: $retryOnFailure,
+        );
+    }
+}

app/modules/Webhooks/Application/WebhookService.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Modules\Webhooks\Application;
+
+use App\Application\Domain\ValueObjects\Uuid;
+use Modules\Webhooks\Domain\WebhookEvent;
+use Modules\Webhooks\Domain\WebhookRepositoryInterface;
+use Modules\Webhooks\Domain\WebhookServiceInterface;
+use Modules\Webhooks\Interfaces\Job\JobPayload;
+use Modules\Webhooks\Interfaces\Job\WebhookHandler;
+use Psr\Log\LoggerInterface;
+use Spiral\Queue\QueueConnectionProviderInterface;
+use Spiral\Queue\QueueInterface;
+
+final readonly class WebhookService implements WebhookServiceInterface
+{
+    private QueueInterface $queue;
+
+    public function __construct(
+        private WebhookRepositoryInterface $webhooks,
+        private LoggerInterface $logger,
+        QueueConnectionProviderInterface $provider,
+    ) {
+        $this->queue = $provider->getConnection('webhook');
+    }
+
+    public function send(WebhookEvent $event): void
+    {
+        $found = $this->webhooks->findByEvent($event->event);
+
+        foreach ($found as $webhook) {
+            $this->sendWebhook($webhook->uuid, $event);
+        }
+    }
+
+    public function sendWebhook(Uuid $uuid, WebhookEvent $event): void
+    {
+        $webhook = $this->webhooks->getByUuid($uuid);
+
+        $this->logger->debug('Sending webhook', ['webhook' => $event->event, 'uuid' => (string) $webhook->uuid]);
+
+        $this->queue->push(
+            WebhookHandler::class,
+            new JobPayload(
+                $webhook->uuid->toObject(),
+                $event->event,
+                $event->payload,
+            ),
+        );
+    }
+}