Merge pull request #175 from City-of-Helsinki/UHF-10559

UHF-10559: Add support for secondary key authentication

documentation/pubsub-messaging.md

@@ -4,10 +4,10 @@ Provides an integration to [Azure's Web PubSub service](https://azure.microsoft.
 
 ## Configuration
 
-You must define a [JSON Vault item](/documentation/api-accounts.md#managing-external-api-credentials) to use this feature. The data field should be a JSON string containing `endpoint`, `hub`, `group` and `access_key`:
+You must define a [JSON Vault item](/documentation/api-accounts.md#managing-external-api-credentials) to use this feature. The data field should be a JSON string containing `endpoint`, `hub`, `group` and `access_key` and optional `secondary_access_key`:
 
 ```json
-{"endpoint": "<endpoint>", "hub": "<hub>", "group": "<group>", "access_key": "<access-key>"}
+{"endpoint": "<endpoint>", "hub": "<hub>", "group": "<group>", "access_key": "<access-key>", "secondary_access_key":  "<secondary-access-key>"}
 ```
 
 ## Usage
@@ -85,34 +85,8 @@ $pubsub_account = [
     'hub' => '<hub>',
     'group' => '<group>',
     'access_key' => '<access-key>',
+    'secondary_access_key' => '<secondary-access-key>',
   ]),
 ];
 $config['helfi_api_base.api_accounts']['vault'][] = $pubsub_account;
 ```
-
-## Solving pubsub related problems
-
-If menus or news or other content doesn't update normally, you can verify that the pubsub service is working correctly
-
-### Artemis is not up on etusivu-instance
-- See that frontpage production has artemis pod up and running
-
-#### If the pod is not running
-- See if there is a pipeline to get it up again OR
-- Contact HiQ
-
-
-### Pubsub-process is not running
-- Go to any production site's cron pod
-  - run `ps aux`, you should see pubsub related process on the list
-
-#### If the process is not running
-- Short term solution is to run `drush cr` to force the site to fetch new data.
-- You can run production deployment to get it running again.
-
-
-### Bad credentials
-- Go to any cron pod and look for authorization error
-
-#### Update the credentials
-- Go and update the pubsub-vault credentials

drush.services.yml

@@ -7,12 +7,6 @@ services:
       - '@datetime.time'
     tags:
       - { name: drush.command }
-  helfi_api_base.pubsub_commands:
-    class: \Drupal\helfi_api_base\Commands\PubSubCommands
-    arguments:
-      - '@helfi_api_base.pubsub_manager'
-    tags:
-      - { name: drush.command }
   helfi_api_base.deploy_commands:
     class: \Drupal\helfi_api_base\Commands\DeployCommands
     arguments: ['@event_dispatcher']

helfi_api_base.services.yml

@@ -108,53 +108,17 @@ services:
     arguments:
       - '@config.factory'
 
-  Drupal\helfi_api_base\Azure\PubSub\SettingsFactory: '@helfi_api_base.pubsub_settings_factory'
-  helfi_api_base.pubsub_settings_factory:
-    class: Drupal\helfi_api_base\Azure\PubSub\SettingsFactory
-    arguments:
-      - '@helfi_api_base.vault_manager'
-
-  Drupal\helfi_api_base\Azure\PubSub\PubSubClientFactory: '@helfi_api_base.pubsub_client_factory'
-  helfi_api_base.pubsub_client_factory:
-    class: Drupal\helfi_api_base\Azure\PubSub\PubSubClientFactory
-
-  helfi_api_base.pubsub_client:
-    public: false
-    class: \WebSocket\Client
-    factory: ['@helfi_api_base.pubsub_client_factory', 'create']
-    arguments:
-      - '@helfi_api_base.pubsub_settings'
-      - '@datetime.time'
-
-  Drupal\helfi_api_base\Azure\PubSub\Settings: '@helfi_api_base.pubsub_settings'
-  helfi_api_base.pubsub_settings:
-    class: Drupal\helfi_api_base\Azure\PubSub\Settings
-    factory: ['@helfi_api_base.pubsub_settings_factory', 'create']
-
-  Drupal\helfi_api_base\Azure\PubSub\PubSubManager: '@helfi_api_base.pubsub_manager'
-  helfi_api_base.pubsub_manager:
-    class: Drupal\helfi_api_base\Azure\PubSub\PubSubManager
-    arguments:
-      - '@helfi_api_base.pubsub_client'
-      - '@event_dispatcher'
-      - '@datetime.time'
-      - '@helfi_api_base.pubsub_settings'
-
-  Drupal\helfi_api_base\Cache\CacheTagInvalidatorInterface: '@helfi_api_base.cache_tag_invalidator'
-  Drupal\helfi_api_base\Cache\CacheTagInvalidator: '@helfi_api_base.cache_tag_invalidator'
-  helfi_api_base.cache_tag_invalidator:
-    class: Drupal\helfi_api_base\Cache\CacheTagInvalidator
-    arguments:
-      - '@helfi_api_base.pubsub_manager'
-
-  Drupal\helfi_api_base\EventSubscriber\CacheTagInvalidatorSubscriber: '@helfi_api_base.cache_tag_invalidator_subscriber'
-  helfi_api_base.cache_tag_invalidator_subscriber:
-    class: Drupal\helfi_api_base\EventSubscriber\CacheTagInvalidatorSubscriber
-    arguments:
-      - '@cache_tags.invalidator'
-      - '@helfi_api_base.environment_resolver'
-    tags:
-      - { name: event_subscriber }
+  Drupal\helfi_api_base\Azure\PubSub\PubSubClientFactory: ~
+  Drupal\helfi_api_base\Azure\PubSub\PubSubClientFactoryInterface: '@Drupal\helfi_api_base\Azure\PubSub\PubSubClientFactory'
+  Drupal\helfi_api_base\Azure\PubSub\SettingsFactory: ~
+  Drupal\helfi_api_base\Azure\PubSub\Settings:
+    factory: ['@Drupal\helfi_api_base\Azure\PubSub\SettingsFactory', 'create']
+  Drupal\helfi_api_base\Azure\PubSub\PubSubManager: ~
+  Drupal\helfi_api_base\Azure\PubSub\PubSubManagerInterface: '@Drupal\helfi_api_base\Azure\PubSub\PubSubManager'
+  Drupal\helfi_api_base\Cache\CacheTagInvalidatorInterface: '@Drupal\helfi_api_base\Cache\CacheTagInvalidator'
+  Drupal\helfi_api_base\Cache\CacheTagInvalidator: ~
+
+  Drupal\helfi_api_base\EventSubscriber\CacheTagInvalidatorSubscriber: ~
 
   Drupal\helfi_api_base\Entity\Revision\RevisionManager: '@helfi_api_base.revision_manager'
   helfi_api_base.revision_manager:

src/Azure/PubSub/PubSubClientFactory.php

@@ -11,31 +11,43 @@
 /**
  * A Web socket client factory.
  */
-final class PubSubClientFactory {
+final class PubSubClientFactory implements PubSubClientFactoryInterface {
 
   /**
-   * Constructs a new websocket client object.
+   * Constructs a new instance.
    *
-   * @param \Drupal\helfi_api_base\Azure\PubSub\Settings $settings
-   *   The settings.
    * @param \Drupal\Component\Datetime\TimeInterface $time
    *   The time interface.
+   * @param \Drupal\helfi_api_base\Azure\PubSub\Settings $settings
+   *   The PubSub settings.
+   */
+  public function __construct(
+    private readonly TimeInterface $time,
+    private readonly Settings $settings,
+  ) {
+  }
+
+  /**
+   * Constructs a new websocket client object.
+   *
+   * @param string $accessKey
+   *   The access key.
    *
    * @return \WebSocket\Client
    *   The client.
    */
-  public function create(Settings $settings, TimeInterface $time) : Client {
-    $url = sprintf('wss://%s/client/hubs/%s', rtrim($settings->endpoint, '/'), $settings->hub);
+  public function create(string $accessKey) : Client {
+    $url = sprintf('wss://%s/client/hubs/%s', rtrim($this->settings->endpoint, '/'), $this->settings->hub);
 
     $authorizationToken = JWT::encode([
       'aud' => $url,
-      'iat' => $time->getCurrentTime(),
-      'exp' => $time->getCurrentTime() + 3600,
+      'iat' => $this->time->getCurrentTime(),
+      'exp' => $this->time->getCurrentTime() + 3600,
       'role' => [
         'webpubsub.sendToGroup',
         'webpubsub.joinLeaveGroup',
       ],
-    ], $settings->accessKey, 'HS256');
+    ], $accessKey, 'HS256');
 
     return new Client($url, [
       'headers' => [

src/Azure/PubSub/PubSubClientFactoryInterface.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Drupal\helfi_api_base\Azure\PubSub;
+
+use WebSocket\Client;
+
+/**
+ * A Web socket client factory.
+ */
+interface PubSubClientFactoryInterface {
+
+  /**
+   * Constructs a new websocket client object.
+   *
+   * @param string $accessKey
+   *   The access key.
+   *
+   * @return \WebSocket\Client
+   *   The client.
+   */
+  public function create(string $accessKey): Client;
+
+}

src/Azure/PubSub/PubSubManager.php

@@ -5,6 +5,9 @@
 namespace Drupal\helfi_api_base\Azure\PubSub;
 
 use Drupal\Component\Datetime\TimeInterface;
+use Drupal\Core\Utility\Error;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\DependencyInjection\Attribute\Autowire;
 use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
 use WebSocket\Client;
 use WebSocket\ConnectionException;
@@ -15,29 +18,32 @@
 final class PubSubManager implements PubSubManagerInterface {
 
   /**
-   * A flag indicating whether we've joined the group.
+   * The websocket client.
    *
-   * @var bool
+   * @var \WebSocket\Client|null
    */
-  private bool $joinedGroup = FALSE;
+  private ?Client $client = NULL;
 
   /**
    * Constructs a new instance.
    *
-   * @param \WebSocket\Client $client
-   *   The websocket client.
+   * @param \Drupal\helfi_api_base\Azure\PubSub\PubSubClientFactoryInterface $clientFactory
+   *   The client factory.
    * @param \Symfony\Contracts\EventDispatcher\EventDispatcherInterface $eventDispatcher
    *   The event dispatcher service.
    * @param \Drupal\Component\Datetime\TimeInterface $time
    *   The datetime service.
    * @param \Drupal\helfi_api_base\Azure\PubSub\Settings $settings
    *   The PubSub settings.
+   * @param \Psr\Log\LoggerInterface $logger
+   *   The logger.
    */
   public function __construct(
-    private readonly Client $client,
+    private readonly PubSubClientFactoryInterface $clientFactory,
     private readonly EventDispatcherInterface $eventDispatcher,
     private readonly TimeInterface $time,
     private readonly Settings $settings,
+    #[Autowire(service: 'logger.channel.helfi_api_base')] private readonly LoggerInterface $logger,
   ) {
   }
 
@@ -48,31 +54,45 @@ public function __construct(
    * @throws \WebSocket\ConnectionException
    * @throws \WebSocket\TimeoutException
    */
-  private function joinGroup() : void {
-    if ($this->joinedGroup) {
+  private function initializeClient() : void {
+    if ($this->client) {
       return;
     }
-    $this->client->text(
-      $this->encodeMessage([
-        'type' => 'joinGroup',
-        'group' => $this->settings->group,
-      ])
-    );
+    $client = $exception = NULL;
+
+    // Initialize client with primary key, fallback to secondary key.
+    foreach ($this->settings->accessKeys as $key) {
+      $exception = NULL;
+
+      try {
+        $client = $this->clientFactory->create($key);
+        $client->text($this->encodeMessage([
+          'type' => 'joinGroup',
+          'group' => $this->settings->group,
+        ]));
+      }
+      catch (ConnectionException $exception) {
+        Error::logException($this->logger, $exception);
+      }
+    }
+
+    if ($exception instanceof ConnectionException) {
+      throw $exception;
+    }
 
     try {
       // Wait until we've actually joined the group.
-      $message = $this->decodeMessage((string) $this->client->receive());
+      $message = $this->decodeMessage((string) $client->receive());
 
       if (isset($message['event']) && $message['event'] === 'connected') {
-        $this->joinedGroup = TRUE;
+        $this->client = $client;
 
         return;
       }
     }
     catch (\JsonException) {
     }
-
-    throw new ConnectionException('Failed to join a group.');
+    throw new ConnectionException('Failed to initialize the client.');
   }
 
   /**
@@ -105,57 +125,28 @@ private function decodeMessage(string $message) : array {
     return json_decode($message, TRUE, flags: JSON_THROW_ON_ERROR);
   }
 
-  /**
-   * {@inheritdoc}
-   */
-  public function setTimeout(int $timeout) : self {
-    $this->client->setTimeout($timeout);
-    return $this;
-  }
-
-  /**
-   * Asserts the settings.
-   *
-   * This is used to exit early if required settings are not populated.
-   */
-  private function assertSettings() : void {
-    $vars = get_object_vars($this->settings);
-
-    foreach ($vars as $key => $value) {
-      if (empty($this->settings->{$key})) {
-        throw new ConnectionException("Azure PubSub '$key' is not configured.");
-      }
-    }
-  }
-
   /**
    * {@inheritdoc}
    */
   public function sendMessage(array $message) : self {
-    $this->assertSettings();
-    $this->joinGroup();
-
-    $this->client
-      ->text(
-        $this->encodeMessage([
-          'type' => 'sendToGroup',
-          'group' => $this->settings->group,
-          'dataType' => 'json',
-          'data' => $message + [
-            'timestamp' => $this->time->getCurrentTime(),
-          ],
-        ])
-      );
-
+    $this->initializeClient();
+
+    $this->client->text($this->encodeMessage([
+      'type' => 'sendToGroup',
+      'group' => $this->settings->group,
+      'dataType' => 'json',
+      'data' => $message + [
+        'timestamp' => $this->time->getCurrentTime(),
+      ],
+    ]));
     return $this;
   }
 
   /**
    * {@inheritdoc}
    */
   public function receive() : string {
-    $this->assertSettings();
-    $this->joinGroup();
+    $this->initializeClient();
 
     $message = (string) $this->client->receive();
     $json = $this->decodeMessage($message);

src/Azure/PubSub/PubSubManagerInterface.php

@@ -36,15 +36,4 @@ public function sendMessage(array $message): self;
    */
   public function receive(): string;
 
-  /**
-   * Sets the client timeout.
-   *
-   * @param int $timeout
-   *   The timeout in seconds.
-   *
-   * @return self
-   *   The self.
-   */
-  public function setTimeout(int $timeout): self;
-
 }