[papi] add watch workspace status api

Author: mustard-mh

components/dashboard/src/service/json-rpc-workspace-client.ts

@@ -4,12 +4,14 @@
  * See License.AGPL.txt in the project root for license information.
  */
 
-import { Code, ConnectError, PromiseClient } from "@connectrpc/connect";
+import { CallOptions, Code, ConnectError, PromiseClient } from "@connectrpc/connect";
 import { PartialMessage } from "@bufbuild/protobuf";
 import { WorkspaceService } from "@gitpod/public-api/lib/gitpod/v1/workspace_connect";
 import { GetWorkspaceRequest, GetWorkspaceResponse } from "@gitpod/public-api/lib/gitpod/v1/workspace_pb";
 import { converter } from "./public-api";
 import { getGitpodService } from "./service";
+import { generateAsyncGenerator } from "@gitpod/gitpod-protocol/lib/generate-async-generator";
+import { WorkspaceInstance } from "@gitpod/gitpod-protocol";
 
 export class JsonRpcWorkspaceClient implements PromiseClient<typeof WorkspaceService> {
     async getWorkspace(request: PartialMessage<GetWorkspaceRequest>): Promise<GetWorkspaceResponse> {
@@ -22,4 +24,47 @@ export class JsonRpcWorkspaceClient implements PromiseClient<typeof WorkspaceSer
         result.item = workspace;
         return result;
     }
+
+    async *watchWorkspaceStatus(
+        request: PartialMessage<WatchWorkspaceStatusRequest>,
+        options?: CallOptions,
+    ): AsyncIterable<WatchWorkspaceStatusResponse> {
+        if (!options?.signal) {
+            throw new ConnectError("signal is required", Code.InvalidArgument);
+        }
+        const it = generateAsyncGenerator<WorkspaceInstance>(
+            (sink) => {
+                const dispose = getGitpodService().registerClient({
+                    onInstanceUpdate: (instance) => {
+                        sink.next(instance);
+                    },
+                });
+                return dispose.dispose;
+            },
+            { signal: options.signal },
+        );
+        if (request.workspaceId) {
+            const resp = await this.getWorkspace({ id: request.workspaceId });
+            if (resp.item?.status) {
+                const response = new WatchWorkspaceStatusResponse();
+                response.status = resp.item.status;
+                yield response;
+            }
+        }
+        for await (const item of it) {
+            if (!item) {
+                continue;
+            }
+            if (request.workspaceId && item.workspaceId !== request.workspaceId) {
+                continue;
+            }
+            const status = converter.toWorkspace(item).status;
+            if (!status) {
+                continue;
+            }
+            const response = new WatchWorkspaceStatusResponse();
+            response.status = status;
+            yield response;
+        }
+    }
 }

components/gitpod-protocol/src/generate-async-generator.ts

@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) 2023 Gitpod GmbH. All rights reserved.
+ * Licensed under the GNU Affero General Public License (AGPL).
+ * See License.AGPL.txt in the project root for license information.
+ */
+
+type Sink<T> = {
+    next: (value: T | void) => void;
+};
+
+/**
+ * Generates an asynchronous generator that yields values based on the provided setup function.
+ *
+ * the setup function that takes a sink and returns a cleanup function.
+ * setup sink object has a `next` method that accepts a value to be pushed to the generator.
+ */
+export async function* generateAsyncGenerator<T>(
+    setup: (sink: Sink<T>) => () => void,
+    opts: { signal: AbortSignal },
+): AsyncGenerator<T | void, void, unknown> {
+    const queue: T[] = [];
+
+    let resolveNext: ((value: T | void) => void) | null = null;
+
+    const sink: Sink<T> = {
+        next: (value: T | void) => {
+            if (resolveNext) {
+                resolveNext(value);
+                resolveNext = null;
+            } else {
+                if (value) {
+                    queue.push(value);
+                }
+            }
+        },
+    };
+
+    let isStopped = false;
+    opts.signal.addEventListener("abort", () => {
+        isStopped = true;
+        sink.next();
+    });
+
+    const cleanup = setup(sink);
+
+    try {
+        while (!isStopped) {
+            if (queue.length) {
+                yield queue.shift();
+            } else {
+                yield new Promise<T | void>((resolve) => {
+                    resolveNext = resolve;
+                });
+            }
+        }
+        // ignore error since code in `try` scope will not throw an error
+        // unless caller use it.throw, then it will throw to itself
+    } finally {
+        cleanup();
+    }
+}

components/public-api/gitpod/v1/workspace.proto

@@ -12,12 +12,29 @@ service WorkspaceService {
   // +return NOT_FOUND User does not have access to a workspace with the given
   // ID +return NOT_FOUND Workspace does not exist
   rpc GetWorkspace(GetWorkspaceRequest) returns (GetWorkspaceResponse) {}
+
+  // WatchWorkspaceStatus watchs the workspaces status changes
+  //
+  // ID +return NOT_FOUND Workspace does not exist
+  rpc WatchWorkspaceStatus(WatchWorkspaceStatusRequest) returns (stream WatchWorkspaceStatusResponse) {}
 }
 
 message GetWorkspaceRequest { string id = 1; }
 
 message GetWorkspaceResponse { Workspace item = 1; }
 
+message WatchWorkspaceStatusRequest {
+  // workspace_id specifies the workspace to watch
+  //
+  // +optional if empty then watch all workspaces
+  optional string workspace_id = 1;
+}
+
+message WatchWorkspaceStatusResponse {
+  // item is the workspace that was changed
+  WorkspaceStatus status = 1;
+}
+
 // +resource get workspace
 message Workspace {
   string id = 1;