actually run bulk ops

Author: jordanverasamy

packages/store/src/cli/commands/store/execute.ts

@@ -4,6 +4,7 @@ import {readFile} from '@shopify/cli-kit/node/fs'
 import {ensureAuthenticatedAdmin} from '@shopify/cli-kit/node/session'
 import {adminRequest} from '@shopify/cli-kit/node/api/admin'
 import {parseGraphQLOperation} from '../../services/graphql-parser.js'
+import {runBulkQuery} from '../../services/bulk-operations.js'
 
 export default class Execute extends Command {
   static summary = 'execute a graphql query or mutation on a store'
@@ -89,8 +90,15 @@ export default class Execute extends Command {
 
     if (flags['bulk-operation']) {
       const operationType = parseGraphQLOperation(query)
-      this.log(`executing bulk ${operationType}...`)
-      this.error('bulk operations not yet implemented')
+
+      if (operationType === 'query') {
+        this.log('starting bulk query...')
+        const results = await runBulkQuery(query, adminSession)
+        this.log(results)
+      } else {
+        this.error('bulk mutations not yet implemented')
+      }
+      return
     }
 
     const result = await adminRequest(query, adminSession, variables)

packages/store/src/cli/services/bulk-operations.ts

@@ -0,0 +1,72 @@
+import {AdminSession} from '@shopify/cli-kit/node/session'
+import {adminRequest} from '@shopify/cli-kit/node/api/admin'
+import {sleep} from '@shopify/cli-kit/node/system'
+import {fetch} from '@shopify/cli-kit/node/http'
+
+const BULK_QUERY_MUTATION = `
+  mutation bulkOperationRunQuery($query: String!) {
+    bulkOperationRunQuery(query: $query) {
+      bulkOperation {
+        id
+        status
+      }
+      userErrors {
+        field
+        message
+      }
+    }
+  }
+`
+
+const BULK_OPERATION_STATUS_QUERY = `
+  query {
+    currentBulkOperation {
+      id
+      status
+      errorCode
+      objectCount
+      url
+    }
+  }
+`
+
+interface BulkOperationStatus {
+  currentBulkOperation: {
+    id: string
+    status: 'CREATED' | 'RUNNING' | 'COMPLETED' | 'FAILED' | 'CANCELED'
+    errorCode?: string
+    objectCount: string
+    url?: string
+  }
+}
+
+export async function runBulkQuery(
+  query: string,
+  session: AdminSession,
+): Promise<string> {
+  const startResult = await adminRequest<any>(BULK_QUERY_MUTATION, session, {query})
+
+  if (startResult.bulkOperationRunQuery.userErrors.length > 0) {
+    const errors = startResult.bulkOperationRunQuery.userErrors.map((e: any) => e.message).join(', ')
+    throw new Error(`bulk operation failed: ${errors}`)
+  }
+
+  while (true) {
+    await sleep(1000)
+
+    const statusResult = await adminRequest<BulkOperationStatus>(BULK_OPERATION_STATUS_QUERY, session)
+    const operation = statusResult.currentBulkOperation
+
+    if (operation.status === 'COMPLETED') {
+      if (!operation.url) {
+        throw new Error('bulk operation completed but no results url')
+      }
+      const response = await fetch(operation.url)
+      return await response.text()
+    }
+
+    if (operation.status === 'FAILED') {
+      throw new Error(`bulk operation failed: ${operation.errorCode}`)
+    }
+  }
+}