From 5d93c28b6416d9dc7794037fe691015c20525343 Mon Sep 17 00:00:00 2001
From: Matthew Bystedt <mbystedt@gmail.com>
Date: Wed, 18 Sep 2024 16:55:46 -0700
Subject: [PATCH] fix: broker account token events (#269)

---
 src/collection/account.service.ts             | 15 ++---
 src/collection/collection.controller.ts       |  2 +-
 src/constants.ts                              |  2 +-
 .../inspector-account.component.ts            |  8 +--
 ui/src/app/service/system-api.service.ts      | 55 +++++++++++--------
 5 files changed, 44 insertions(+), 38 deletions(-)

diff --git a/src/collection/account.service.ts b/src/collection/account.service.ts
index 3a1c8e21..54dd7403 100644
--- a/src/collection/account.service.ts
+++ b/src/collection/account.service.ts
@@ -211,6 +211,12 @@ export class AccountService {
       ['token', 'generated'],
     );
 
+    this.redisService.publish(REDIS_PUBSUB.BROKER_ACCOUNT_TOKEN, {
+      data: {
+        clientId: account.clientId,
+      },
+    });
+
     return {
       token,
     };
@@ -275,15 +281,6 @@ export class AccountService {
         await this.addTokenToServiceTools(projectName, serviceName, {
           [`broker-jwt:${account.clientId}`]: token,
         });
-        this.redisService.publish(REDIS_PUBSUB.VAULT_SERVICE_TOKEN, {
-          data: {
-            clientId: account.clientId,
-            environment: 'tools',
-            project: projectName,
-            service: serviceName,
-            scmUrl: service.collection.scmUrl,
-          },
-        });
       } catch (err) {
         // Log?
       }
diff --git a/src/collection/collection.controller.ts b/src/collection/collection.controller.ts
index 729929ac..721dec8c 100644
--- a/src/collection/collection.controller.ts
+++ b/src/collection/collection.controller.ts
@@ -173,7 +173,7 @@ export class CollectionController {
   @UseGuards(BrokerCombinedAuthGuard)
   @ApiBearerAuth()
   tokenUpdatedEvents(): Observable<MessageEvent> {
-    return this.redis.getEventSource(REDIS_PUBSUB.VAULT_SERVICE_TOKEN);
+    return this.redis.getEventSource(REDIS_PUBSUB.BROKER_ACCOUNT_TOKEN);
   }
 
   @Get('service/:id/details')
diff --git a/src/constants.ts b/src/constants.ts
index 77d33832..4f8f4a25 100644
--- a/src/constants.ts
+++ b/src/constants.ts
@@ -94,7 +94,7 @@ export const OAUTH2_CLIENT_MAP_USERNAME =
 
 export const REDIS_PUBSUB = {
   GRAPH: 'graph',
-  VAULT_SERVICE_TOKEN: 'vault-service-token',
+  BROKER_ACCOUNT_TOKEN: 'broker-account-token',
 } as const;
 
 export const GITHUB_CLIENT_ID = process.env.GITHUB_CLIENT_ID ?? '';
diff --git a/ui/src/app/graph/inspector-account/inspector-account.component.ts b/ui/src/app/graph/inspector-account/inspector-account.component.ts
index 6050b899..506fc8df 100644
--- a/ui/src/app/graph/inspector-account/inspector-account.component.ts
+++ b/ui/src/app/graph/inspector-account/inspector-account.component.ts
@@ -71,9 +71,9 @@ export class InspectorAccountComponent implements OnChanges, OnInit, OnDestroy {
       .createAccountTokenEventSource()
       .subscribe({
         next: (data: any) => {
-          console.log(data);
-          // TODO: only update if necessary
-          this.updateAccount();
+          if (data.clientId === this.account.clientId) {
+            this.updateAccount();
+          }
         },
         error: (error: any) => {
           console.error('Error receiving token update events:', error);
@@ -105,7 +105,7 @@ export class InspectorAccountComponent implements OnChanges, OnInit, OnDestroy {
       .afterClosed()
       .subscribe(() => {
         this.requestedAccountId = undefined;
-        this.updateAccount();
+        //this.updateAccount();
       });
   }
 
diff --git a/ui/src/app/service/system-api.service.ts b/ui/src/app/service/system-api.service.ts
index 016cf8a8..ef15f28c 100644
--- a/ui/src/app/service/system-api.service.ts
+++ b/ui/src/app/service/system-api.service.ts
@@ -1,6 +1,6 @@
 import { HttpClient } from '@angular/common/http';
 import { Injectable } from '@angular/core';
-import { filter, map, Observable } from 'rxjs';
+import { filter, finalize, map, Observable, share } from 'rxjs';
 import { SseClient } from 'ngx-sse-client';
 
 import { environment } from '../../environments/environment';
@@ -16,6 +16,8 @@ export class SystemApiService {
     private sseClient: SseClient,
   ) {}
 
+  static accountEventObserver: Observable<any> | null = null;
+
   getAccountTokens(accountId: string) {
     return this.http.get<JwtRegistryDto[]>(
       `${environment.apiUrl}/v1/collection/broker-account/${accountId}/token`,
@@ -52,29 +54,36 @@ export class SystemApiService {
   }
 
   createAccountTokenEventSource(): Observable<any> {
-    return this.sseClient
-      .stream(`${environment.apiUrl}/v1/graph/token-updated`)
-      .pipe(
-        filter((event) => {
-          if (event.type === 'error') {
-            const errorEvent = event as ErrorEvent;
-            if (errorEvent.error) {
-              console.error(errorEvent.error, errorEvent.message);
+    if (!SystemApiService.accountEventObserver) {
+      SystemApiService.accountEventObserver = this.sseClient
+        .stream(`${environment.apiUrl}/v1/collection/broker-account/events`)
+        .pipe(
+          filter((event) => {
+            if (event.type === 'error') {
+              const errorEvent = event as ErrorEvent;
+              if (errorEvent.error) {
+                console.error(errorEvent.error, errorEvent.message);
+              }
+              return false;
+            }
+            return true;
+          }),
+          map((event) => {
+            if (event.type !== 'error') {
+              const messageEvent = event as MessageEvent;
+              //console.info(
+              //  `SSE request with type "${messageEvent.type}" and data "${messageEvent.data}"`,
+              //);
+              return JSON.parse(messageEvent.data);
             }
-            return false;
-          }
-          return true;
-        }),
-        map((event) => {
-          if (event.type !== 'error') {
-            const messageEvent = event as MessageEvent;
-            //console.info(
-            //  `SSE request with type "${messageEvent.type}" and data "${messageEvent.data}"`,
-            //);
-            return JSON.parse(messageEvent.data);
-          }
-        }),
-      );
+          }),
+          finalize(() => {
+            SystemApiService.accountEventObserver = null;
+          }),
+          share(),
+        );
+    }
+    return SystemApiService.accountEventObserver;
   }
 
   getConnectionConfig() {