[server] Broadcast workspace instance update to listeners

Author: easyCZ

components/server/src/messaging/redis-subscriber.ts

@@ -22,10 +22,14 @@ import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/expe
 import { Attributes } from "@gitpod/gitpod-protocol/lib/experiments/types";
 import { reportRedisUpdateCompleted, reportRedisUpdateReceived } from "../prometheus-metrics";
 import { Redis } from "ioredis";
+import { WorkspaceDB } from "@gitpod/gitpod-db/lib";
 
 @injectable()
 export class RedisSubscriber implements LocalMessageBroker {
-    constructor(@inject(Redis) private readonly redis: Redis) {}
+    constructor(
+        @inject(Redis) private readonly redis: Redis,
+        @inject(WorkspaceDB) private readonly workspaceDB: WorkspaceDB,
+    ) {}
 
     protected workspaceInstanceUpdateListeners: Map<string, WorkspaceInstanceUpdateListener[]> = new Map();
 
@@ -73,6 +77,33 @@ export class RedisSubscriber implements LocalMessageBroker {
 
     private async onInstanceUpdate(update: RedisWorkspaceInstanceUpdate): Promise<void> {
         log.debug("[redis] Received instance update", { update });
+
+        if (!update.ownerID || !update.instanceID) {
+            return;
+        }
+
+        const listeners = this.workspaceInstanceUpdateListeners.get(update.ownerID) || [];
+        if (listeners.length === 0) {
+            return;
+        }
+
+        const ctx = {};
+        const instance = await this.workspaceDB.findInstanceById(update.instanceID);
+        if (!instance) {
+            return;
+        }
+
+        for (const l of listeners) {
+            try {
+                l(ctx, instance);
+            } catch (err) {
+                log.error(
+                    { userId: update.ownerID, instanceId: instance.id, workspaceId: update.workspaceID },
+                    "Failed to broadcast workspace instance update.",
+                    err,
+                );
+            }
+        }
     }
 
     async stop() {