Update notifications on updates to subscription

AlexTugarev · AlexTugarev · commit f72360117b76 · 2022-09-13T12:23:51.000Z
diff --git a/components/dashboard/src/AppNotifications.tsx b/components/dashboard/src/AppNotifications.tsx
@@ -19,15 +19,22 @@ export function AppNotifications() {
             setNotifications(localState);
             return;
         }
-        (async () => {
-            const serverState = await getGitpodService().server.getNotifications();
-            setNotifications(serverState);
-            if (serverState.length > 0) {
-                setLocalStorageObject(KEY_APP_NOTIFICATIONS, serverState, /* expires in */ 60 /* seconds */);
-            }
-        })();
+        reloadNotifications().catch(console.error);
+
+        getGitpodService().registerClient({
+            onSubscriptionUpdate: () => reloadNotifications().catch(console.error),
+        });
     }, []);
 
+    const reloadNotifications = async () => {
+        const serverState = await getGitpodService().server.getNotifications();
+        setNotifications(serverState);
+        removeLocalStorageObject(KEY_APP_NOTIFICATIONS);
+        if (serverState.length > 0) {
+            setLocalStorageObject(KEY_APP_NOTIFICATIONS, serverState, /* expires in */ 300 /* seconds */);
+        }
+    };
+
     const topNotification = notifications[0];
     if (topNotification === undefined) {
         return null;
diff --git a/components/gitpod-messagebus/src/messagebus.ts b/components/gitpod-messagebus/src/messagebus.ts
@@ -61,6 +61,8 @@ export interface MessageBusHelper {
      * @param topic the topic to parse
      */
     getWsInformationFromTopic(topic: string): WorkspaceTopic | undefined;
+
+    getSubscriptionUpdateTopic(attributionId?: string): string;
 }
 
 export const WorkspaceTopic = Symbol("WorkspaceTopic");
@@ -89,6 +91,10 @@ export class MessageBusHelperImpl implements MessageBusHelper {
         await ch.assertExchange(this.workspaceExchange, "topic", { durable: true });
     }
 
+    getSubscriptionUpdateTopic(attributionId: string | undefined): string {
+        return `subscription.${attributionId || "*"}.update`;
+    }
+
     /**
      * Computes the topic name of for listening to a workspace.
      *
diff --git a/components/gitpod-protocol/src/gitpod-service.ts b/components/gitpod-protocol/src/gitpod-service.ts
@@ -70,6 +70,8 @@ export interface GitpodClient {
 
     onPrebuildUpdate(update: PrebuildWithStatus): void;
 
+    onSubscriptionUpdate(attributionId: string): void;
+
     onCreditAlert(creditAlert: CreditAlert): void;
 
     //#region propagating reconnection to iframe
@@ -571,6 +573,18 @@ export class GitpodCompositeClient<Client extends GitpodClient> implements Gitpo
             }
         }
     }
+
+    onSubscriptionUpdate(attributionId: string): void {
+        for (const client of this.clients) {
+            if (client.onSubscriptionUpdate) {
+                try {
+                    client.onSubscriptionUpdate(attributionId);
+                } catch (error) {
+                    console.error(error);
+                }
+            }
+        }
+    }
 }
 
 export type GitpodService = GitpodServiceImpl<GitpodClient, GitpodServer>;
diff --git a/components/server/ee/src/workspace/gitpod-server-impl.ts b/components/server/ee/src/workspace/gitpod-server-impl.ts
@@ -115,6 +115,7 @@ import { BillingModes } from "../billing/billing-mode";
 import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server";
 import { BillingService } from "../billing/billing-service";
 import Stripe from "stripe";
+import { MessageBusIntegration } from "../../../src/workspace/messagebus-integration";
 
 @injectable()
 export class GitpodServerEEImpl extends GitpodServerImpl {
@@ -162,6 +163,8 @@ export class GitpodServerEEImpl extends GitpodServerImpl {
     @inject(BillingModes) protected readonly billingModes: BillingModes;
     @inject(BillingService) protected readonly billingService: BillingService;
 
+    @inject(MessageBusIntegration) protected readonly messageBus: MessageBusIntegration;
+
     initialize(
         client: GitpodClient | undefined,
         user: User | undefined,
@@ -174,6 +177,7 @@ export class GitpodServerEEImpl extends GitpodServerImpl {
 
         this.listenToCreditAlerts();
         this.listenForPrebuildUpdates().catch((err) => log.error("error registering for prebuild updates", err));
+        this.listenForSubscriptionUpdates().catch((err) => log.error("error registering for prebuild updates", err));
     }
 
     protected async listenForPrebuildUpdates() {
@@ -201,6 +205,32 @@ export class GitpodServerEEImpl extends GitpodServerImpl {
         // TODO(at) we need to keep the list of accessible project up to date
     }
 
+    protected async listenForSubscriptionUpdates() {
+        if (!this.user) {
+            return;
+        }
+        const teamIds = (await this.teamDB.findTeamsByUser(this.user.id)).map(({ id }) =>
+            AttributionId.render({ kind: "team", teamId: id }),
+        );
+        for (const attributionId of [AttributionId.render({ kind: "user", userId: this.user.id }), ...teamIds]) {
+            this.disposables.push(
+                this.localMessageBroker.listenForSubscriptionUpdates(
+                    attributionId,
+                    (ctx: TraceContext, attributionId: AttributionId) =>
+                        TraceContext.withSpan(
+                            "forwardSubscriptionUpdateToClient",
+                            (ctx) => {
+                                traceClientMetadata(ctx, this.clientMetadata);
+                                TraceContext.setJsonRPCMetadata(ctx, "onSubscriptionUpdate");
+                                this.client?.onSubscriptionUpdate(AttributionId.render(attributionId));
+                            },
+                            ctx,
+                        ),
+                ),
+            );
+        }
+    }
+
     protected async getAccessibleProjects() {
         if (!this.user) {
             return [];
@@ -2154,6 +2184,9 @@ export class GitpodServerEEImpl extends GitpodServerImpl {
                 spendingLimit: this.defaultSpendingLimit,
                 billingStrategy: "stripe",
             });
+
+            console.log("notifyOnSubscriptionUpdate");
+            this.messageBus.notifyOnSubscriptionUpdate(ctx, attributionId).catch();
         } catch (error) {
             log.error(`Failed to subscribe '${attributionId}' to Stripe`, error);
             throw new ResponseError(
@@ -2241,6 +2274,9 @@ export class GitpodServerEEImpl extends GitpodServerImpl {
             spendingLimit: usageLimit,
             billingStrategy: costCenter?.billingStrategy || "other",
         });
+
+        console.log("notifyOnSubscriptionUpdate");
+        this.messageBus.notifyOnSubscriptionUpdate(ctx, attributionId).catch();
     }
 
     async getNotifications(ctx: TraceContext): Promise<string[]> {
diff --git a/components/server/src/messaging/local-message-broker.ts b/components/server/src/messaging/local-message-broker.ts
@@ -12,6 +12,7 @@ import {
     WorkspaceInstance,
 } from "@gitpod/gitpod-protocol";
 import { CreditAlert } from "@gitpod/gitpod-protocol/lib/accounting-protocol";
+import { AttributionId } from "@gitpod/gitpod-protocol/lib/attribution";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { inject, injectable } from "inversify";
@@ -20,6 +21,9 @@ import { MessageBusIntegration } from "../workspace/messagebus-integration";
 export interface PrebuildUpdateListener {
     (ctx: TraceContext, evt: PrebuildWithStatus): void;
 }
+export interface SubscriptionUpdateListener {
+    (ctx: TraceContext, attributionId: AttributionId): void;
+}
 export interface CreditAlertListener {
     (ctx: TraceContext, alert: CreditAlert): void;
 }
@@ -38,6 +42,8 @@ export interface LocalMessageBroker {
 
     listenForPrebuildUpdates(projectId: string, listener: PrebuildUpdateListener): Disposable;
 
+    listenForSubscriptionUpdates(attributionId: string, listener: SubscriptionUpdateListener): Disposable;
+
     listenToCreditAlerts(userId: string, listener: CreditAlertListener): Disposable;
 
     listenForPrebuildUpdatableEvents(listener: HeadlessWorkspaceEventListener): Disposable;
@@ -69,6 +75,7 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
     protected creditAlertsListeners: Map<string, CreditAlertListener[]> = new Map();
     protected headlessWorkspaceEventListeners: Map<string, HeadlessWorkspaceEventListener[]> = new Map();
     protected workspaceInstanceUpdateListeners: Map<string, WorkspaceInstanceUpdateListener[]> = new Map();
+    protected subscriptionUpdateListeners: Map<string, SubscriptionUpdateListener[]> = new Map();
 
     protected readonly disposables = new DisposableCollection();
 
@@ -151,6 +158,24 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
                 },
             ),
         );
+        this.disposables.push(
+            this.messageBusIntegration.listenToSubscriptionUpdates(
+                (ctx: TraceContext, attributionId: AttributionId) => {
+                    TraceContext.setOWI(ctx, {});
+
+                    const listeners = this.subscriptionUpdateListeners.get(AttributionId.render(attributionId)) || [];
+
+                    for (const l of listeners) {
+                        try {
+                            l(ctx, attributionId);
+                        } catch (err) {
+                            TraceContext.setError(ctx, err);
+                            log.error("listenForWorkspaceInstanceUpdates", err, { attributionId });
+                        }
+                    }
+                },
+            ),
+        );
     }
 
     async stop() {
@@ -165,6 +190,10 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
         return this.doRegister(userId, listener, this.creditAlertsListeners);
     }
 
+    listenForSubscriptionUpdates(attributionId: string, listener: SubscriptionUpdateListener): Disposable {
+        return this.doRegister(attributionId, listener, this.subscriptionUpdateListeners);
+    }
+
     listenForPrebuildUpdatableEvents(listener: HeadlessWorkspaceEventListener): Disposable {
         // we're being cheap here in re-using a map where it just needs to be a plain array.
         return this.doRegister(
diff --git a/components/server/src/workspace/messagebus-integration.ts b/components/server/src/workspace/messagebus-integration.ts
@@ -22,6 +22,7 @@ import * as opentracing from "opentracing";
 import { CancellationTokenSource } from "vscode-ws-jsonrpc";
 import { increaseMessagebusTopicReads } from "../prometheus-metrics";
 import { CreditAlert } from "@gitpod/gitpod-protocol/lib/accounting-protocol";
+import { AttributionId } from "@gitpod/gitpod-protocol/lib/attribution";
 
 interface WorkspaceInstanceUpdateCallback {
     (ctx: TraceContext, instance: WorkspaceInstance, ownerId: string | undefined): void;
@@ -72,6 +73,16 @@ export class CreditAlertListener extends AbstractTopicListener<CreditAlert> {
     }
 }
 
+export class SubscriptionUpdateListener extends AbstractTopicListener<AttributionId> {
+    constructor(protected messageBusHelper: MessageBusHelper, listener: TopicListener<AttributionId>) {
+        super(messageBusHelper.workspaceExchange, listener);
+    }
+
+    topic() {
+        return this.messageBusHelper.getSubscriptionUpdateTopic();
+    }
+}
+
 export class PrebuildUpdatableQueueListener implements MessagebusListener {
     protected channel: Channel | undefined;
     protected consumerTag: string | undefined;
@@ -208,6 +219,28 @@ export class MessageBusIntegration extends AbstractMessageBusIntegration {
         return Disposable.create(() => cancellationTokenSource.cancel());
     }
 
+    async notifyOnSubscriptionUpdate(ctx: TraceContext, attributionId: AttributionId) {
+        if (!this.channel) {
+            throw new Error("Not connected to message bus");
+        }
+        const topic = this.messageBusHelper.getSubscriptionUpdateTopic(AttributionId.render(attributionId));
+        const msg = Buffer.from(JSON.stringify(attributionId));
+        await this.messageBusHelper.assertWorkspaceExchange(this.channel);
+
+        await super.publish(MessageBusHelperImpl.WORKSPACE_EXCHANGE_LOCAL, topic, msg, {
+            trace: ctx,
+        });
+    }
+
+    listenToSubscriptionUpdates(callback: (ctx: TraceContext, attributionId: AttributionId) => void): Disposable {
+        const listener = new SubscriptionUpdateListener(this.messageBusHelper, callback);
+        const cancellationTokenSource = new CancellationTokenSource();
+        this.listen(listener, cancellationTokenSource.token).catch((err) => {
+            /** ignore */
+        });
+        return Disposable.create(() => cancellationTokenSource.cancel());
+    }
+
     async notifyOnPrebuildUpdate(prebuildInfo: PrebuildWithStatus) {
         if (!this.channel) {
             throw new Error("Not connected to message bus");

Original file line number	Diff line number	Diff line change
`@@ -61,6 +61,8 @@ export interface MessageBusHelper {`
`61`	`61`	`* @param topic the topic to parse`
`62`	`62`	`*/`
`63`	`63`	`getWsInformationFromTopic(topic: string): WorkspaceTopic \| undefined;`
	`64`	`+`
	`65`	`+ getSubscriptionUpdateTopic(attributionId?: string): string;`
`64`	`66`	`}`
`65`	`67`
`66`	`68`	`export const WorkspaceTopic = Symbol("WorkspaceTopic");`
`@@ -89,6 +91,10 @@ export class MessageBusHelperImpl implements MessageBusHelper {`
`89`	`91`	`await ch.assertExchange(this.workspaceExchange, "topic", { durable: true });`
`90`	`92`	`}`
`91`	`93`
	`94`	`+ getSubscriptionUpdateTopic(attributionId: string \| undefined): string {`
	`95`	+ return `subscription.${attributionId \|\| "*"}.update`;
	`96`	`+ }`
	`97`	`+`
`92`	`98`	`/**`
`93`	`99`	`* Computes the topic name of for listening to a workspace.`
`94`	`100`	`*`
Original file line number	Diff line number	Diff line change
`@@ -70,6 +70,8 @@ export interface GitpodClient {`
`70`	`70`
`71`	`71`	`onPrebuildUpdate(update: PrebuildWithStatus): void;`
`72`	`72`
	`73`	`+ onSubscriptionUpdate(attributionId: string): void;`
	`74`	`+`
`73`	`75`	`onCreditAlert(creditAlert: CreditAlert): void;`
`74`	`76`
`75`	`77`	`//#region propagating reconnection to iframe`
`@@ -571,6 +573,18 @@ export class GitpodCompositeClient<Client extends GitpodClient> implements Gitpo`
`571`	`573`	`}`
`572`	`574`	`}`
`573`	`575`	`}`
	`576`	`+`
	`577`	`+ onSubscriptionUpdate(attributionId: string): void {`
	`578`	`+ for (const client of this.clients) {`
	`579`	`+ if (client.onSubscriptionUpdate) {`
	`580`	`+ try {`
	`581`	`+ client.onSubscriptionUpdate(attributionId);`
	`582`	`+ } catch (error) {`
	`583`	`+ console.error(error);`
	`584`	`+ }`
	`585`	`+ }`
	`586`	`+ }`
	`587`	`+ }`
`574`	`588`	`}`
`575`	`589`
`576`	`590`	`export type GitpodService = GitpodServiceImpl<GitpodClient, GitpodServer>;`