fix: broker account token events (#269)

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?
  }

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')

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 ?? '';

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();
  });
  }
 

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() {