wip

Author: danigargar

asterisk/agi/src/Agi/Webhook/WebhookEventPublisher.php

@@ -18,6 +18,10 @@ public function __construct(
 
     public function publish(string $event): void
     {
+        // Disabled: the realtime webhook dispatcher (subscribed to Kamailio
+        // pubsub) is now the single source of webhook events.
+        return;
+
         try {
             $brandId = (int) $this->agi->getVariable('BRANDID');
             if ($brandId === 0) {

microservices/workers/bin/realtime-webhook-dispatcher

@@ -0,0 +1,21 @@
+#!/usr/bin/env php
+<?php
+
+use Symfony\Component\Console\Input\ArgvInput;
+use Symfony\Component\HttpFoundation\Request;
+
+require __DIR__.'/../config/bootstrap.php';
+
+$input = new ArgvInput();
+$env = $input->getParameterOption(['--env', '-e'], getenv('APP_ENV') ?: 'dev');
+
+$kernel = new Kernel($env, false);
+$request = new Request([], [], [], [], [], ['REQUEST_URI' => '/realtime-webhook-dispatcher']);
+
+$response = $kernel->handle($request);
+$response->send();
+$kernel->terminate($request, $response);
+
+if ($response->getStatusCode() >= 300) {
+    die(1);
+}

microservices/workers/config/routes.yaml

@@ -36,4 +36,8 @@ asterisk-hint-update:
 
 webhook-dispatcher:
   path: /webhook-dispatcher
-  controller: Worker\Webhooks::dispatch
+  controller: Worker\Webhooks::dispatch
+
+realtime-webhook-dispatcher:
+  path: /realtime-webhook-dispatcher
+  controller: Worker\RealtimeWebhooks::dispatch

microservices/workers/config/services.yaml

@@ -74,3 +74,8 @@ services:
             $redisDb: '%redis_jobs_db%'
             $redisTimeout: '%redis_timeout%'
 
+    Worker\RealtimeWebhooks:
+        public: true
+        arguments:
+            $logger: '@monolog.logger.workers'
+

microservices/workers/src/Worker/RealtimeWebhooks.php

@@ -0,0 +1,120 @@
+<?php
+
+namespace Worker;
+
+use Ivoz\Core\Infrastructure\Persistence\Redis\RedisMasterFactory;
+use Ivoz\Provider\Domain\Job\WebhookJobInterface;
+use Ivoz\Provider\Domain\Model\Webhook\Payload\WebhookEventPayload;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\HttpFoundation\Response;
+
+class RealtimeWebhooks
+{
+    private const REDIS_REALTIME_DB = 1;
+    private const CHANNEL_PATTERN = 'users:*';
+
+    private const EVENT_MAP = [
+        'Trying' => 'start',
+        'Proceeding' => 'ring',
+        'Early' => 'ring',
+        'Confirmed' => 'answer',
+        'Terminated' => 'end',
+    ];
+
+    /** @var array<string, array{Party: ?string, Direction: ?string, Owner: ?string}> */
+    private array $callCache = [];
+
+    public function __construct(
+        private RedisMasterFactory $redisMasterFactory,
+        private WebhookJobInterface $webhookJob,
+        private LoggerInterface $logger,
+    ) {
+    }
+
+    public function dispatch(): Response
+    {
+        $redis = $this->redisMasterFactory->create(self::REDIS_REALTIME_DB);
+
+        $redis->pSubscribe(
+            [self::CHANNEL_PATTERN],
+            function ($redis, $pattern, $channel, $message): void {
+                try {
+                    $this->processMessage($channel, $message);
+                } catch (\Throwable $e) {
+                    $this->logger->error(
+                        '[RT-WEBHOOK] Error processing message on ' . $channel . ': ' . $e->getMessage()
+                    );
+                }
+            }
+        );
+
+        return new Response('', 500);
+    }
+
+    private function processMessage(string $channel, string $message): void
+    {
+        if (!preg_match('/^users:b(\d+):c(\d+):/', $channel, $matches)) {
+            return;
+        }
+
+        $brandId = (int) $matches[1];
+        $companyId = (int) $matches[2];
+
+        /** @var array<string, mixed>|null $data */
+        $data = json_decode($message, true);
+        if (!is_array($data) || !isset($data['Event'])) {
+            return;
+        }
+
+        $kamEvent = (string) $data['Event'];
+        $callId = isset($data['Call-ID']) ? (string) $data['Call-ID'] : null;
+
+        if ($kamEvent === 'UpdateCLID') {
+            return;
+        }
+
+        $webhookEvent = self::EVENT_MAP[$kamEvent] ?? null;
+        if ($webhookEvent === null) {
+            return;
+        }
+
+        if ($kamEvent === 'Trying' && $callId !== null) {
+            $this->callCache[$callId] = [
+                'Party' => isset($data['Party']) ? (string) $data['Party'] : null,
+                'Direction' => isset($data['Direction']) ? (string) $data['Direction'] : null,
+                'Owner' => isset($data['Owner']) ? (string) $data['Owner'] : null,
+            ];
+        }
+
+        $cached = ($callId !== null && isset($this->callCache[$callId]))
+            ? $this->callCache[$callId]
+            : ['Party' => null, 'Direction' => null, 'Owner' => null];
+
+        $party = isset($data['Party']) ? (string) $data['Party'] : $cached['Party'];
+        $direction = $cached['Direction'];
+
+        $payload = new WebhookEventPayload(
+            event: $webhookEvent,
+            brandId: $brandId,
+            companyId: $companyId,
+            ddiId: null,
+            ddiE164: null,
+            callId: $callId,
+            uniqueId: isset($data['ID']) ? (string) $data['ID'] : null,
+            caller: ($direction === 'inbound') ? $party : null,
+            callee: ($direction === 'outbound') ? $party : null,
+            dialStatus: null,
+            timestamp: isset($data['Time'])
+                ? date('Y-m-d\TH:i:s.v\Z', (int) $data['Time'])
+                : null,
+        );
+
+        $this->webhookJob
+            ->setData($payload)
+            ->send();
+
+        if ($kamEvent === 'Terminated' && $callId !== null) {
+            unset($this->callCache[$callId]);
+        }
+    }
+}

profiles/portal/etc/supervisor/conf.d/realtime-webhook-daemon.conf

@@ -0,0 +1,9 @@
+[program:realtime-webhook-dispatcher]
+command=/usr/bin/php /opt/irontec/ivozprovider/microservices/workers/bin/realtime-webhook-dispatcher
+autorestart=true
+autostart=true
+user=www-data
+startretries=5
+process_name=%(program_name)s
+numprocs=1
+stopsignal=KILL