Shopify Bulk Operation API で 10 万件商品を高速同期する設計

mutation {
  bulkOperationRunQuery(
    query: """
    {
      products {
        edges {
          node {
            id
            title
            updatedAt
            variants {
              edges {
                node {
                  id
                  sku
                  price
                  inventoryQuantity
                }
              }
            }
          }
        }
      }
    }
    """
  ) {
    bulkOperation {
      id
      status
    }
    userErrors {
      field
      message
    }
  }
}

async function pollBulkOperation(jobId: string): Promise<string> {
  const query = `
    query {
      currentBulkOperation {
        id
        status
        errorCode
        objectCount
        url
      }
    }
  `
 
  let delay = 5_000 // 最初は 5 秒待つ
  const MAX_DELAY = 120_000 // 最大 2 分
 
  while (true) {
    await sleep(delay)
 
    const { data } = await shopify.graphql(query)
    const op = data.currentBulkOperation
 
    // 別のジョブが割り込んでいる場合は id が変わる
    if (op.id !== jobId) {
      throw new Error(`Unexpected job in progress: ${op.id}`)
    }
 
    if (op.status === "COMPLETED") {
      if (!op.url) throw new Error("COMPLETED but url is null")
      return op.url
    }
 
    if (op.status === "FAILED" || op.status === "CANCELED") {
      throw new Error(`Bulk operation ${op.status}: ${op.errorCode}`)
    }
 
    // RUNNING / CANCELING → 待機継続
    delay = Math.min(delay * 1.5, MAX_DELAY) // 指数バックオフ
  }
}
 
function sleep(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms))
}

import { createInterface } from "readline"
import { Readable } from "stream"
 
interface ProductNode {
  id: string
  title: string
  updatedAt: string
}
 
interface VariantNode {
  id: string
  sku: string
  price: string
  inventoryQuantity: number
  __parentId: string // 親 Product の id
}
 
async function processBulkResult(url: string): Promise<void> {
  const response = await fetch(url)
  if (!response.ok || !response.body) {
    throw new Error(`Failed to fetch JSONL: ${response.status}`)
  }
 
  const productMap = new Map<string, ProductNode>()
  const rl = createInterface({
    input: Readable.fromWeb(response.body as any),
    crlfDelay: Infinity,
  })
 
  for await (const line of rl) {
    if (!line.trim()) continue
    const obj = JSON.parse(line)
 
    if (obj.__parentId) {
      // Variant 行 → 親 Product と紐づけて処理
      const variant = obj as VariantNode
      const product = productMap.get(variant.__parentId)
      if (product) {
        await upsertVariant(product, variant)
      }
    } else {
      // Product 行 → Map に保持（後続 Variant のために）
      const product = obj as ProductNode
      productMap.set(product.id, product)
    }
  }
}

// Step 1: Staged Upload を作成（POST 形式で受け取る）
const stagedUploadMutation = `
  mutation {
    stagedUploadsCreate(input: {
      resource: BULK_MUTATION_VARIABLES
      filename: "variants.jsonl"
      mimeType: "text/jsonl"
      httpMethod: POST
    }) {
      stagedTargets {
        url
        parameters { name value }
      }
      userErrors { field message }
    }
  }
`
const { data: uploadData } = await shopify.graphql(stagedUploadMutation)
const target = uploadData.stagedUploadsCreate.stagedTargets[0]
const params = target.parameters as { name: string; value: string }[]
 
// Step 2: 返ってきた parameters とファイルを multipart/form-data で POST
const jsonl = variantsToUpsert
  .map((v) =>
    JSON.stringify({
      input: {
        id: v.shopifyVariantId,
        price: v.price,
        inventoryQuantities: {
          availableQuantity: v.quantity,
          locationId: LOCATION_GID,
        },
      },
    })
  )
  .join("\n")
 
const form = new FormData()
// parameters を先に append し、file は必ず最後に append する
for (const param of params) {
  form.append(param.name, param.value)
}
form.append("file", new Blob([jsonl], { type: "text/jsonl" }), "variants.jsonl")
await fetch(target.url, { method: "POST", body: form })
 
// Step 3: ジョブ登録
// stagedUploadPath には resourceUrl ではなく、parameters の "key"
// （アップロード済みオブジェクトのパス）を渡す
const stagedUploadPath = params.find((p) => p.name === "key")!.value
 
const bulkMutation = `
  mutation bulkRunMutation($stagedUploadPath: String!) {
    bulkOperationRunMutation(
      mutation: "mutation updateVariant($input: ProductVariantInput!) { productVariantUpdate(input: $input) { productVariant { id price } userErrors { field message } } }"
      stagedUploadPath: $stagedUploadPath
    ) {
      bulkOperation { id status }
      userErrors { field message }
    }
  }
`
await shopify.graphql(bulkMutation, { stagedUploadPath })

async function upsertVariant(product: ProductNode, variant: VariantNode) {
  await db
    .insertInto("variants")
    .values({
      sku: variant.sku,
      shopify_variant_id: variant.id,
      shopify_product_id: product.id,
      price: variant.price,
      inventory_quantity: variant.inventoryQuantity,
      synced_at: new Date().toISOString(),
    })
    .onConflict((oc) =>
      oc.column("sku").doUpdateSet({
        shopify_variant_id: (eb) => eb.ref("excluded.shopify_variant_id"),
        price: (eb) => eb.ref("excluded.price"),
        inventory_quantity: (eb) => eb.ref("excluded.inventory_quantity"),
        synced_at: (eb) => eb.ref("excluded.synced_at"),
      })
    )
    .execute()
}