Check worker status before calling success callback

PR-URL: https://github.com/hasura/graphql-engine-mono/pull/10070
GitOrigin-RevId: e25c9fb5b89c4463766eca0909120d57b2b4c14f

frontend/libs/console/legacy-ce/src/lib/components/Services/Data/DataSources/CreateDataSource/Neon/useCreateHasuraCloudDatasource.ts

@@ -16,6 +16,7 @@ import Globals from '../../../../../../Globals';
 import {
   controlPlaneClient,
   FETCH_CONFIG_STATUS,
+  FetchConfigStatusSubscription,
 } from '../../../../../../features/ControlPlane';
 
 type HasuraDBCreationPayload = {
@@ -162,46 +163,58 @@ export function useCreateHasuraCloudDatasource(
            * If the change has not propagated to a worker to which the `pg_add_source` request is routed to,
            * the server would throw an `inconsistent_object` error as the new env var will not be found.
            *
-           * To handle this, we wait for the changes to be propagated to all
+           * To handle this, we wait for the changes to be propagated to all the live hasura
            * workers of a tenant before attempting to connect the DB.
            *
            * We do that by verifying that the new config hash has been set
-           * for all workers of the tenant using the `config_status` table.
+           * for all the live workers of the tenant using the `config_status` table
+           *
+           * Hash is compared against the `config_status` table and worker status
+           * is checked against the `hasura_worker` table
            *
            * Additionally add a timeout interval with retries as an additional
            * redundancy since it throws CORS error locally
            */
-          const { unsubscribe } = controlPlaneClient.subscribe(
-            FETCH_CONFIG_STATUS,
-            {
-              tenantId: Globals.hasuraCloudTenantId,
-            },
-            data => {
-              if (
-                // check if all workers are successfully configured with the new hash
-                data.config_status.every(
-                  (config: { hash: string; message: string }) =>
-                    config.message === 'Service configured successfully' &&
-                    config.hash !== oldConfigHash
-                )
-              ) {
-                verifyProjectHealthAndConnectDataSource(
-                  successCallback,
-                  errorCallback
-                );
+          const { unsubscribe } =
+            controlPlaneClient.subscribe<FetchConfigStatusSubscription>(
+              FETCH_CONFIG_STATUS,
+              {
+                tenantId: Globals.hasuraCloudTenantId,
+              },
+              data => {
+                const configStatusData = data.hasura_worker
+                  .map(config_status_obj =>
+                    config_status_obj.config_statuses.map(
+                      config_data => config_data
+                    )
+                  )
+                  .flat(1);
+
+                if (
+                  // check if all workers are successfully configured with the new hash
+                  configStatusData.every(
+                    config =>
+                      config.message === 'Service configured successfully' &&
+                      config.hash !== oldConfigHash
+                  )
+                ) {
+                  verifyProjectHealthAndConnectDataSource(
+                    successCallback,
+                    errorCallback
+                  );
+                  unsubscribe();
+                }
+              },
+              error => {
+                programmaticallyTraceError({
+                  error:
+                    'failed subscribing to fetch_config_status while connecting neon database',
+                  cause: error,
+                });
+                errorCallback();
                 unsubscribe();
               }
-            },
-            error => {
-              programmaticallyTraceError({
-                error:
-                  'failed subscribing to fetch_config_status while connecting neon database',
-                cause: error,
-              });
-              errorCallback();
-              unsubscribe();
-            }
-          );
+            );
         })
         .catch(error => {
           // if adding env var fails unexpectedly, set the error state

frontend/libs/console/legacy-ce/src/lib/components/Services/Data/DataSources/CreateDataSource/utils.ts

@@ -94,7 +94,7 @@ export const updateEnvVars = (
         if (response.errors) {
           throw new Error(response.errors[0]?.message);
         } else {
-          return response.updateTenantEnv;
+          return response.data.updateTenantEnv;
         }
       });
     })

frontend/libs/console/legacy-ce/src/lib/features/ControlPlane/queries.ts

@@ -265,15 +265,12 @@ export const ADD_SCHEMA_REGISTRY_FEATURE_REQUEST = gql(`
 `);
 
 export const FETCH_CONFIG_STATUS = gql(`
-  subscription FetchConfigStatus($tenantId: uuid!) {
-    config_status(
-      where: {
-        tenant_id: { _eq: $tenantId }
-        is_active: { _eq: true }
-      }
-    ) {
+subscription FetchConfigStatus($tenantId: uuid!) {
+  hasura_worker(where: {config_statuses: {is_active: {_eq: true}, tenant_id: {_eq: $tenantId}}, status: {_eq: "live"}}) {
+    config_statuses(where: {is_active: {_eq: true}, tenant_id: {_eq: $tenantId}}) {
       hash
       message
+      worker_id
     }
   }
-`);
+}`);