feat: controlpanel move to a accept first strategy for batch

cli/src/commands/subgraph/commands/batch-publish.ts

@@ -10,6 +10,7 @@ import { getBaseHeaders } from '../../../core/config.js';
 import { handleCompositionResult } from '../../../handle-composition-result.js';
 import { limitMaxValue } from '../../../constants.js';
 import { fileExists } from '../../../utils.js';
+import { pollBatchPublishStatus } from '../utils/poll-batch-publish-status.js';
 
 const entrySchema = z.object({
   name: z.string().trim().min(1, 'a non-empty "name" is required'),
@@ -57,6 +58,7 @@ export default (opts: BaseCommandOptions) => {
   );
   command.option('-r, --raw', 'Prints to the console in json format instead of table');
   command.option('-j, --json', 'Prints to the console in json format instead of table');
+  command.option('--async', 'This flag enables polling the operation status');
 
   command.action(async (options) => {
     const configFile = resolve(options.config);
@@ -126,18 +128,34 @@ export default (opts: BaseCommandOptions) => {
       spinner.start();
     }
 
-    const resp = await opts.client.platform.publishFederatedSubgraphs(
+    let resp = await opts.client.platform.publishFederatedSubgraphs(
       {
         namespace: options.namespace,
         subgraphs,
         disableResolvabilityValidation: options.disableResolvabilityValidation,
         limit,
+        async: options.async,
       },
       {
         headers: getBaseHeaders(),
       },
     );
 
+    if (resp.jobId) {
+      const controller = new AbortController();
+      function cancelPolling() {
+        controller.abort();
+      }
+
+      process.on('SIGINT', cancelPolling);
+
+      try {
+        resp = await pollBatchPublishStatus(opts.client, resp.jobId, controller.signal);
+      } finally {
+        process.off('SIGINT', cancelPolling);
+      }
+    }
+
     const total = subgraphs.length;
     const changed = resp.updatedSubgraphNames?.length ?? 0;
 

cli/src/commands/subgraph/utils/poll-batch-publish-status.ts

@@ -0,0 +1,83 @@
+import {
+  PublishFederatedSubgraphsResponse,
+  BatchPublishJobStatus,
+} from '@wundergraph/cosmo-connect/dist/platform/v1/platform_pb';
+import { EnumStatusCode } from '@wundergraph/cosmo-connect/dist/common/common_pb';
+import { Client } from '../../../core/client/client.js';
+import { getBaseHeaders } from '../../../core/config.js';
+
+export async function pollBatchPublishStatus(
+  client: Client,
+  jobId: string,
+  signal: AbortSignal,
+): Promise<PublishFederatedSubgraphsResponse> {
+  let attempt = 0;
+  const headers = getBaseHeaders();
+  while (!signal.aborted) {
+    const resp = await client.platform.getBatchPublishJobStatus({ jobId }, { headers, signal });
+    if (resp.response?.code !== EnumStatusCode.OK) {
+      return new PublishFederatedSubgraphsResponse({
+        response: resp.response,
+      });
+    }
+
+    switch (resp.status) {
+      case BatchPublishJobStatus.PENDING:
+      case BatchPublishJobStatus.PROCESSING: {
+        await sleep(computeDelay(1000, 5000, attempt++, true), signal);
+        break;
+      }
+      case BatchPublishJobStatus.FAILED: {
+        return new PublishFederatedSubgraphsResponse({
+          response: {
+            code: EnumStatusCode.ERR_SUBGRAPH_COMPOSITION_FAILED,
+            details: resp.failureReason,
+          },
+        });
+      }
+      case BatchPublishJobStatus.COMPLETED: {
+        return new PublishFederatedSubgraphsResponse({
+          response: { code: EnumStatusCode.OK },
+          ...resp,
+        });
+      }
+    }
+  }
+
+  /**
+   * The only reason we should realistically get here is due to `signal` being aborted; however, we still need
+   * to return a response object
+   */
+  return new PublishFederatedSubgraphsResponse({
+    response: {
+      code: EnumStatusCode.ERR,
+      details: signal.aborted ? 'Operation was cancelled by the user.' : undefined,
+    },
+  });
+}
+
+function computeDelay(base: number, max: number, attempt: number, jitter: boolean): number {
+  const delay = Math.min(max, base * 2 ** attempt);
+  return jitter ? delay * (0.5 + Math.random() * 0.5) : delay;
+}
+
+function sleep(ms: number, signal?: AbortSignal): Promise<'aborted' | 'ok'> {
+  return new Promise((resolve) => {
+    if (signal?.aborted) {
+      resolve('aborted');
+      return;
+    }
+
+    const timer = setTimeout(() => {
+      signal?.removeEventListener('abort', onAbort);
+      resolve('ok');
+    }, ms);
+
+    function onAbort() {
+      clearTimeout(timer);
+      resolve('aborted');
+    }
+
+    signal?.addEventListener('abort', onAbort, { once: true });
+  });
+}