[dashboard] enable grpc migration per service (#19140)

components/dashboard/src/service/public-api.ts

@@ -46,30 +46,50 @@ export const converter = new PublicAPIConverter();
 
 export const helloService = createPromiseClient(HelloService, transport);
 export const personalAccessTokensService = createPromiseClient(TokensService, transport);
+/**
+ * @deprecated use configurationClient instead
+ */
 export const projectsService = createPromiseClient(ProjectsService, transport);
 /**
  * @deprecated use workspaceClient instead
  */
 export const workspacesService = createPromiseClient(WorkspaceV1Service, transport);
 export const oidcService = createPromiseClient(OIDCService, transport);
 
-export const workspaceClient = createServiceClient(WorkspaceService, new JsonRpcWorkspaceClient());
-export const organizationClient = createServiceClient(
-    OrganizationService,
-    new JsonRpcOrganizationClient(),
-    "organization",
-);
+export const workspaceClient = createServiceClient(WorkspaceService, {
+    client: new JsonRpcWorkspaceClient(),
+    featureFlagSuffix: "workspace",
+});
+export const organizationClient = createServiceClient(OrganizationService, {
+    client: new JsonRpcOrganizationClient(),
+    featureFlagSuffix: "organization",
+});
 // No jsonrcp client for the configuration service as it's only used in new UI of the dashboard
 export const configurationClient = createServiceClient(ConfigurationService);
-export const prebuildClient = createServiceClient(PrebuildService, new JsonRpcPrebuildClient());
+export const prebuildClient = createServiceClient(PrebuildService, {
+    client: new JsonRpcPrebuildClient(),
+    featureFlagSuffix: "prebuild",
+});
 
-export const authProviderClient = createServiceClient(AuthProviderService, new JsonRpcAuthProviderClient());
+export const authProviderClient = createServiceClient(AuthProviderService, {
+    client: new JsonRpcAuthProviderClient(),
+    featureFlagSuffix: "authprovider",
+});
 
-export const scmClient = createServiceClient(SCMService, new JsonRpcScmClient());
+export const scmClient = createServiceClient(SCMService, {
+    client: new JsonRpcScmClient(),
+    featureFlagSuffix: "scm",
+});
 
-export const envVarClient = createServiceClient(EnvironmentVariableService, new JsonRpcEnvvarClient());
+export const envVarClient = createServiceClient(EnvironmentVariableService, {
+    client: new JsonRpcEnvvarClient(),
+    featureFlagSuffix: "envvar",
+});
 
-export const sshClient = createServiceClient(SSHService, new JsonRpcSSHClient());
+export const sshClient = createServiceClient(SSHService, {
+    client: new JsonRpcSSHClient(),
+    featureFlagSuffix: "ssh",
+});
 
 export async function listAllProjects(opts: { orgId: string }): Promise<ProtocolProject[]> {
     let pagination = {
@@ -128,22 +148,24 @@ export function updateUser(newUser: User | undefined) {
 
 function createServiceClient<T extends ServiceType>(
     type: T,
-    jsonRpcClient?: PromiseClient<T>,
-    featureFlagSuffix?: string,
+    jsonRpcOptions?: {
+        client: PromiseClient<T>;
+        featureFlagSuffix: string;
+    },
 ): PromiseClient<T> {
     return new Proxy(createPromiseClient(type, transport), {
         get(grpcClient, prop) {
             const experimentsClient = getExperimentsClient();
             // TODO(ak) remove after migration
             async function resolveClient(): Promise<PromiseClient<T>> {
-                if (!jsonRpcClient) {
+                if (!jsonRpcOptions) {
                     return grpcClient;
                 }
-                const featureFlags = ["dashboard_public_api_enabled", "centralizedPermissions"];
-                if (featureFlagSuffix) {
-                    featureFlags.push(`dashboard_public_api_${featureFlagSuffix}_enabled`);
-                }
-                // TODO(ak): is not going to work for getLoggedInUser itself
+                const featureFlags = [
+                    "dashboard_public_api_enabled",
+                    `dashboard_public_api_${jsonRpcOptions.featureFlagSuffix}_enabled`,
+                    "centralizedPermissions",
+                ];
                 const resolvedFlags = await Promise.all(
                     featureFlags.map((ff) =>
                         experimentsClient.getValueAsync(ff, false, {
@@ -155,7 +177,7 @@ function createServiceClient<T extends ServiceType>(
                 if (resolvedFlags.every((f) => f === true)) {
                     return grpcClient;
                 }
-                return jsonRpcClient;
+                return jsonRpcOptions.client;
             }
             /**
              * The original application error is retained using gRPC metadata to ensure that existing error handling remains intact.

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

@@ -21,7 +21,7 @@ import { ApplicationError, ErrorCodes } from "./messaging/error";
 export function generateAsyncGenerator<T>(
     setup: (queue: Queue<T>) => (() => void) | void,
     opts: { signal: AbortSignal },
-) {
+): AsyncIterable<T> {
     return new EventIterator<T>((queue) => {
         opts.signal.addEventListener("abort", () => {
             queue.fail(new ApplicationError(ErrorCodes.CANCELLED, "cancelled"));

components/server/src/api/prebuild-service-api.ts

@@ -24,7 +24,7 @@ import { ProjectsService } from "../projects/projects-service";
 import { PrebuildManager } from "../prebuilds/prebuild-manager";
 import { validate as uuidValidate } from "uuid";
 import { ApplicationError, ErrorCodes } from "@gitpod/gitpod-protocol/lib/messaging/error";
-import { ctxUserId } from "../util/request-context";
+import { ctxSignal, ctxUserId } from "../util/request-context";
 import { UserService } from "../user/user-service";
 
 @injectable()
@@ -120,7 +120,9 @@ export class PrebuildServiceAPI implements ServiceImpl<typeof PrebuildServiceInt
             });
             configurationId = resp.prebuild!.configurationId;
         }
-        const it = this.prebuildManager.watchPrebuildStatus(ctxUserId(), configurationId);
+        const it = await this.prebuildManager.watchPrebuildStatus(ctxUserId(), configurationId, {
+            signal: ctxSignal(),
+        });
         for await (const pb of it) {
             if (request.scope.case === "prebuildId") {
                 if (pb.info.id !== request.scope.value) {

components/server/src/api/server.ts

@@ -55,6 +55,8 @@ import { BearerAuth } from "../auth/bearer-authenticator";
 import { ScmServiceAPI } from "./scm-service-api";
 import { SCMService } from "@gitpod/public-api/lib/gitpod/v1/scm_connect";
 import { SSHService } from "@gitpod/public-api/lib/gitpod/v1/ssh_connect";
+import { PrebuildServiceAPI } from "./prebuild-service-api";
+import { PrebuildService } from "@gitpod/public-api/lib/gitpod/v1/prebuild_connect";
 
 decorate(injectable(), PublicAPIConverter);
 
@@ -81,6 +83,7 @@ export class API {
     @inject(Config) private readonly config: Config;
     @inject(UserService) private readonly userService: UserService;
     @inject(BearerAuth) private readonly bearerAuthenticator: BearerAuth;
+    @inject(PrebuildServiceAPI) private readonly prebuildServiceApi: PrebuildServiceAPI;
 
     listenPrivate(): http.Server {
         const app = express();
@@ -129,6 +132,7 @@ export class API {
                         service(EnvironmentVariableService, this.envvarServiceApi),
                         service(SCMService, this.scmServiceAPI),
                         service(SSHService, this.sshServiceApi),
+                        service(PrebuildService, this.prebuildServiceApi),
                     ]) {
                         router.service(type, new Proxy(impl, this.interceptService(type)));
                     }
@@ -384,6 +388,7 @@ export class API {
         bind(ScmServiceAPI).toSelf().inSingletonScope();
         bind(SSHServiceAPI).toSelf().inSingletonScope();
         bind(StatsServiceAPI).toSelf().inSingletonScope();
+        bind(PrebuildServiceAPI).toSelf().inSingletonScope();
         bind(API).toSelf().inSingletonScope();
     }
 }

components/server/src/api/teams.spec.db.ts

@@ -33,6 +33,7 @@ import { ScmService } from "../scm/scm-service";
 import { ContextService } from "../workspace/context-service";
 import { ContextParser } from "../workspace/context-parser-service";
 import { SSHKeyService } from "../user/sshkey-service";
+import { PrebuildManager } from "../prebuilds/prebuild-manager";
 
 const expect = chai.expect;
 
@@ -63,6 +64,7 @@ export class APITeamsServiceSpec {
         this.container.bind(ContextService).toConstantValue({} as ContextService);
         this.container.bind(ContextParser).toConstantValue({} as ContextParser);
         this.container.bind(SSHKeyService).toConstantValue({} as SSHKeyService);
+        this.container.bind(PrebuildManager).toConstantValue({} as PrebuildManager);
 
         // Clean-up database
         const typeorm = testContainer.get<TypeORM>(TypeORM);

components/server/src/prebuilds/prebuild-manager.ts

@@ -39,7 +39,6 @@ import { ContextParser } from "../workspace/context-parser-service";
 import { IAnalyticsWriter } from "@gitpod/gitpod-protocol/lib/analytics";
 import { generateAsyncGenerator } from "@gitpod/gitpod-protocol/lib/generate-async-generator";
 import { RedisSubscriber } from "../messaging/redis-subscriber";
-import { ctxSignal } from "../util/request-context";
 
 export interface StartPrebuildParams {
     user: User;
@@ -80,29 +79,28 @@ export class PrebuildManager {
         return undefined;
     }
 
-    public async *watchPrebuildStatus(userId: string, configurationId: string): AsyncGenerator<PrebuildWithStatus> {
+    public async watchPrebuildStatus(
+        userId: string,
+        configurationId: string,
+        opts: { signal: AbortSignal },
+    ): Promise<AsyncIterable<PrebuildWithStatus>> {
         await this.auth.checkPermissionOnProject(userId, "read_prebuild", configurationId);
-        return generateAsyncGenerator<PrebuildWithStatus>(
-            (sink) => {
-                try {
-                    const toDispose = this.subscriber.listenForPrebuildUpdates(configurationId, (_ctx, prebuild) => {
-                        sink.push(prebuild);
-                    });
-                    return () => {
-                        toDispose.dispose();
-                    };
-                } catch (e) {
-                    if (e instanceof Error) {
-                        sink.fail(e);
-                    } else {
-                        sink.fail(new Error(String(e) || "unknown"));
-                    }
+        return generateAsyncGenerator<PrebuildWithStatus>((sink) => {
+            try {
+                const toDispose = this.subscriber.listenForPrebuildUpdates(configurationId, (_ctx, prebuild) => {
+                    sink.push(prebuild);
+                });
+                return () => {
+                    toDispose.dispose();
+                };
+            } catch (e) {
+                if (e instanceof Error) {
+                    sink.fail(e);
+                } else {
+                    sink.fail(new Error(String(e) || "unknown"));
                 }
-            },
-            {
-                signal: ctxSignal(),
-            },
-        );
+            }
+        }, opts);
     }
 
     async triggerPrebuild(ctx: TraceContext, user: User, projectId: string, branchName: string | null) {

components/server/src/workspace/workspace-service.ts

@@ -6,7 +6,6 @@
 
 import { inject, injectable } from "inversify";
 import * as grpc from "@grpc/grpc-js";
-import { EventIterator } from "event-iterator";
 import { RedisPublisher, WorkspaceDB } from "@gitpod/gitpod-db/lib";
 import {
     GetWorkspaceTimeoutResult,
@@ -757,7 +756,7 @@ export class WorkspaceService {
         return urls;
     }
 
-    public watchWorkspaceStatus(userId: string, opts: { signal: AbortSignal }): EventIterator<WorkspaceInstance> {
+    public watchWorkspaceStatus(userId: string, opts: { signal: AbortSignal }): AsyncIterable<WorkspaceInstance> {
         return generateAsyncGenerator<WorkspaceInstance>((sink) => {
             try {
                 const dispose = this.subscriber.listenForWorkspaceInstanceUpdates(userId, (_ctx, instance) => {