Shopify Webhook完整开发指南

// 创建Webhook的完整示例
async function createWebhook(shop, accessToken, webhookData) {
  const webhook = {
    webhook: {
      topic: webhookData.topic,
      address: webhookData.address,
      format: 'json',
      fields: webhookData.fields || null,
      metafield_namespaces: webhookData.metafieldNamespaces || null,
      private_metafield_namespaces: webhookData.privateMetafieldNamespaces || null
    }
  }
 
  try {
    const response = await fetch(`https://${shop}.myshopify.com/admin/api/2023-10/webhooks.json`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-Shopify-Access-Token': accessToken
      },
      body: JSON.stringify(webhook)
    })
 
    const result = await response.json()
    
    if (!response.ok) {
      throw new Error(`Failed to create webhook: ${JSON.stringify(result.errors)}`)
    }
 
    console.log('Webhook created successfully:', result.webhook)
    return result.webhook
  } catch (error) {
    console.error('Error creating webhook:', error)
    throw error
  }
}
 
// 批量创建常用Webhook
async function setupCommonWebhooks(shop, accessToken, baseUrl) {
  const webhooks = [
    {
      topic: 'orders/create',
      address: `${baseUrl}/webhooks/orders/create`,
      fields: ['id', 'order_number', 'email', 'total_price', 'line_items']
    },
    {
      topic: 'orders/updated',
      address: `${baseUrl}/webhooks/orders/updated`
    },
    {
      topic: 'orders/paid',
      address: `${baseUrl}/webhooks/orders/paid`
    },
    {
      topic: 'products/update',
      address: `${baseUrl}/webhooks/products/update`
    },
    {
      topic: 'customers/create',
      address: `${baseUrl}/webhooks/customers/create`
    },
    {
      topic: 'app/uninstalled',
      address: `${baseUrl}/webhooks/app/uninstalled`
    }
  ]
 
  const createdWebhooks = []
  
  for (const webhookData of webhooks) {
    try {
      const webhook = await createWebhook(shop, accessToken, webhookData)
      createdWebhooks.push(webhook)
      
      // 添加延迟避免API限制
      await new Promise(resolve => setTimeout(resolve, 100))
    } catch (error) {
      console.error(`Failed to create webhook for ${webhookData.topic}:`, error)
    }
  }
 
  return createdWebhooks
}

# GraphQL Mutation创建Webhook
mutation webhookSubscriptionCreate($topic: WebhookSubscriptionTopic!, $webhookSubscription: WebhookSubscriptionInput!) {
  webhookSubscriptionCreate(topic: $topic, webhookSubscription: $webhookSubscription) {
    webhookSubscription {
      id
      callbackUrl
      format
      includedFields
      metafieldNamespaces
    }
    userErrors {
      field
      message
    }
  }
}

// 使用GraphQL创建Webhook
async function createWebhookGraphQL(shop, accessToken, topic, callbackUrl) {
  const mutation = `
    mutation webhookSubscriptionCreate($topic: WebhookSubscriptionTopic!, $webhookSubscription: WebhookSubscriptionInput!) {
      webhookSubscriptionCreate(topic: $topic, webhookSubscription: $webhookSubscription) {
        webhookSubscription {
          id
          callbackUrl
          format
          includedFields
        }
        userErrors {
          field
          message
        }
      }
    }
  `
 
  const variables = {
    topic: topic.toUpperCase().replace('/', '_'),
    webhookSubscription: {
      callbackUrl: callbackUrl,
      format: 'JSON'
    }
  }
 
  const response = await fetch(`https://${shop}.myshopify.com/admin/api/2023-10/graphql.json`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Shopify-Access-Token': accessToken
    },
    body: JSON.stringify({
      query: mutation,
      variables: variables
    })
  })
 
  const result = await response.json()
  
  if (result.data.webhookSubscriptionCreate.userErrors.length > 0) {
    throw new Error(`GraphQL errors: ${JSON.stringify(result.data.webhookSubscriptionCreate.userErrors)}`)
  }
 
  return result.data.webhookSubscriptionCreate.webhookSubscription
}

const crypto = require('crypto')
 
// Webhook验证中间件
function verifyShopifyWebhook(secret) {
  return (req, res, next) => {
    // 获取Shopify发送的HMAC签名
    const hmacHeader = req.get('X-Shopify-Hmac-Sha256')
    
    if (!hmacHeader) {
      return res.status(401).json({ error: 'Missing HMAC signature' })
    }
 
    // 获取原始请求体
    const rawBody = req.body
    
    // 计算期望的HMAC
    const expectedHmac = crypto
      .createHmac('sha256', secret)
      .update(rawBody, 'utf8')
      .digest('base64')
 
    // 比较签名
    if (hmacHeader !== expectedHmac) {
      console.error('HMAC verification failed:', {
        received: hmacHeader,
        expected: expectedHmac
      })
      return res.status(401).json({ error: 'Invalid HMAC signature' })
    }
 
    // 验证通过，继续处理
    next()
  }
}
 
// Express应用示例
const express = require('express')
const app = express()
 
// 重要：使用raw middleware保持原始请求体
app.use('/webhooks', express.raw({ type: 'application/json' }))
 
// 应用验证中间件
app.use('/webhooks', verifyShopifyWebhook(process.env.SHOPIFY_WEBHOOK_SECRET))

// 更全面的安全验证
class WebhookSecurityValidator {
  constructor(webhookSecret) {
    this.webhookSecret = webhookSecret
    this.requestHistory = new Map() // 防重放攻击
  }
 
  // 验证HMAC签名
  verifyHMAC(body, signature) {
    const expectedSignature = crypto
      .createHmac('sha256', this.webhookSecret)
      .update(body, 'utf8')
      .digest('base64')
    
    return signature === expectedSignature
  }
 
  // 验证请求来源
  verifyOrigin(req) {
    const shopDomain = req.get('X-Shopify-Shop-Domain')
    const topic = req.get('X-Shopify-Topic')
    
    // 验证shop域名格式
    if (!shopDomain || !shopDomain.endsWith('.myshopify.com')) {
      return false
    }
 
    // 验证topic格式
    if (!topic || !/^[\w\/]+$/.test(topic)) {
      return false
    }
 
    return true
  }
 
  // 防重放攻击
  preventReplay(req) {
    const timestamp = req.get('X-Shopify-Webhook-Id')
    const currentTime = Date.now()
    const requestTime = parseInt(timestamp, 10)
 
    // 检查时间戳是否在合理范围内（5分钟）
    if (Math.abs(currentTime - requestTime) > 5 * 60 * 1000) {
      return false
    }
 
    // 检查是否已处理过此请求
    if (this.requestHistory.has(timestamp)) {
      return false
    }
 
    // 记录此请求
    this.requestHistory.set(timestamp, currentTime)
    
    // 清理过期记录
    this.cleanupRequestHistory()
    
    return true
  }
 
  cleanupRequestHistory() {
    const cutoffTime = Date.now() - 10 * 60 * 1000 // 10分钟前
    
    for (const [id, time] of this.requestHistory.entries()) {
      if (time < cutoffTime) {
        this.requestHistory.delete(id)
      }
    }
  }
 
  // 完整验证
  validateRequest(req) {
    const signature = req.get('X-Shopify-Hmac-Sha256')
    
    // 1. 验证HMAC签名
    if (!this.verifyHMAC(req.body, signature)) {
      throw new Error('Invalid HMAC signature')
    }
 
    // 2. 验证请求来源
    if (!this.verifyOrigin(req)) {
      throw new Error('Invalid request origin')
    }
 
    // 3. 防重放攻击
    if (!this.preventReplay(req)) {
      throw new Error('Replay attack detected')
    }
 
    return true
  }
}
 
// 使用示例
const validator = new WebhookSecurityValidator(process.env.SHOPIFY_WEBHOOK_SECRET)
 
app.use('/webhooks', (req, res, next) => {
  try {
    validator.validateRequest(req)
    next()
  } catch (error) {
    console.error('Webhook validation failed:', error.message)
    res.status(401).json({ error: error.message })
  }
})

// 订单创建处理
app.post('/webhooks/orders/create', async (req, res) => {
  try {
    const order = JSON.parse(req.body)
    
    console.log(`New order received: ${order.order_number}`)
    
    // 1. 发送订单确认邮件
    await sendOrderConfirmationEmail(order)
    
    // 2. 更新库存系统
    await updateInventorySystem(order.line_items)
    
    // 3. 创建发货任务
    await createShippingTask(order)
    
    // 4. 通知客服团队
    if (order.total_price > 500) {
      await notifyCustomerService(order)
    }
    
    // 5. 同步到ERP系统
    await syncToERP(order)
    
    // 6. 记录订单分析数据
    await recordOrderAnalytics(order)
    
    res.status(200).send('Order processed successfully')
  } catch (error) {
    console.error('Error processing order webhook:', error)
    res.status(500).send('Error processing order')
  }
})
 
// 订单支付完成处理
app.post('/webhooks/orders/paid', async (req, res) => {
  try {
    const order = JSON.parse(req.body)
    
    console.log(`Payment confirmed for order: ${order.order_number}`)
    
    // 1. 更新订单状态
    await updateOrderStatus(order.id, 'paid')
    
    // 2. 触发自动发货流程
    await triggerFulfillmentProcess(order)
    
    // 3. 发送支付确认通知
    await sendPaymentConfirmation(order)
    
    // 4. 更新客户积分
    await updateCustomerLoyaltyPoints(order.customer.id, order.total_price)
    
    // 5. 检查是否需要风险审核
    if (await needsRiskReview(order)) {
      await flagForRiskReview(order)
    }
    
    res.status(200).send('Payment processed successfully')
  } catch (error) {
    console.error('Error processing payment webhook:', error)
    res.status(500).send('Error processing payment')
  }
})
 
// 订单取消处理
app.post('/webhooks/orders/cancelled', async (req, res) => {
  try {
    const order = JSON.parse(req.body)
    
    console.log(`Order cancelled: ${order.order_number}`)
    
    // 1. 恢复库存
    await restoreInventory(order.line_items)
    
    // 2. 处理退款
    if (order.financial_status === 'paid') {
      await processRefund(order)
    }
    
    // 3. 取消发货
    await cancelShipping(order)
    
    // 4. 发送取消确认邮件
    await sendCancellationEmail(order)
    
    // 5. 更新分析数据
    await updateCancellationAnalytics(order)
    
    res.status(200).send('Cancellation processed successfully')
  } catch (error) {
    console.error('Error processing cancellation webhook:', error)
    res.status(500).send('Error processing cancellation')
  }
})

// 产品更新处理
app.post('/webhooks/products/update', async (req, res) => {
  try {
    const product = JSON.parse(req.body)
    
    console.log(`Product updated: ${product.title}`)
    
    // 1. 同步到外部系统
    await syncProductToExternalSystems(product)
    
    // 2. 更新搜索索引
    await updateSearchIndex(product)
    
    // 3. 检查价格变化
    const priceChanges = await detectPriceChanges(product)
    if (priceChanges.length > 0) {
      await handlePriceChanges(product, priceChanges)
    }
    
    // 4. 更新推荐系统
    await updateRecommendationEngine(product)
    
    // 5. 通知相关团队
    if (product.status === 'draft') {
      await notifyContentTeam(product)
    }
    
    res.status(200).send('Product update processed successfully')
  } catch (error) {
    console.error('Error processing product update webhook:', error)
    res.status(500).send('Error processing product update')
  }
})
 
// 产品创建处理
app.post('/webhooks/products/create', async (req, res) => {
  try {
    const product = JSON.parse(req.body)
    
    console.log(`New product created: ${product.title}`)
    
    // 1. 初始化产品数据
    await initializeProductData(product)
    
    // 2. 设置默认SEO
    await setupDefaultSEO(product)
    
    // 3. 创建营销活动
    await createProductMarketingCampaign(product)
    
    // 4. 通知营销团队
    await notifyMarketingTeam(product)
    
    res.status(200).send('Product creation processed successfully')
  } catch (error) {
    console.error('Error processing product creation webhook:', error)
    res.status(500).send('Error processing product creation')
  }
})

// 客户注册处理
app.post('/webhooks/customers/create', async (req, res) => {
  try {
    const customer = JSON.parse(req.body)
    
    console.log(`New customer registered: ${customer.email}`)
    
    // 1. 发送欢迎邮件
    await sendWelcomeEmail(customer)
    
    // 2. 创建客户档案
    await createCustomerProfile(customer)
    
    // 3. 设置默认偏好
    await setupDefaultPreferences(customer)
    
    // 4. 分配客服代表
    await assignCustomerServiceRep(customer)
    
    // 5. 加入营销列表
    await addToMarketingList(customer)
    
    // 6. 创建首购优惠券
    await createFirstPurchaseCoupon(customer)
    
    res.status(200).send('Customer creation processed successfully')
  } catch (error) {
    console.error('Error processing customer creation webhook:', error)
    res.status(500).send('Error processing customer creation')
  }
})

class WebhookProcessor {
  constructor() {
    this.maxRetries = 3
    this.retryDelay = 1000 // 1秒
    this.processingQueue = []
  }
 
  async processWebhook(topic, data, attemptCount = 1) {
    try {
      await this.executeHandler(topic, data)
      console.log(`Webhook processed successfully: ${topic}`)
    } catch (error) {
      console.error(`Webhook processing failed (attempt ${attemptCount}):`, error)
      
      if (attemptCount < this.maxRetries) {
        // 计算退避延迟 (指数退避)
        const delay = this.retryDelay * Math.pow(2, attemptCount - 1)
        
        console.log(`Retrying in ${delay}ms...`)
        
        setTimeout(() => {
          this.processWebhook(topic, data, attemptCount + 1)
        }, delay)
      } else {
        // 达到最大重试次数，记录到死信队列
        await this.handleFailedWebhook(topic, data, error)
      }
    }
  }
 
  async executeHandler(topic, data) {
    const handlers = {
      'orders/create': this.handleOrderCreate.bind(this),
      'orders/updated': this.handleOrderUpdate.bind(this),
      'orders/paid': this.handleOrderPaid.bind(this),
      'products/update': this.handleProductUpdate.bind(this),
      'customers/create': this.handleCustomerCreate.bind(this)
    }
 
    const handler = handlers[topic]
    if (!handler) {
      throw new Error(`No handler found for topic: ${topic}`)
    }
 
    await handler(data)
  }
 
  async handleFailedWebhook(topic, data, error) {
    // 记录到数据库供后续分析
    await this.logFailedWebhook({
      topic,
      data,
      error: error.message,
      timestamp: new Date(),
      attempts: this.maxRetries
    })
 
    // 发送警报通知
    await this.sendFailureAlert(topic, data, error)
  }
 
  // 示例处理器
  async handleOrderCreate(order) {
    // 验证数据完整性
    this.validateOrderData(order)
    
    // 处理订单
    await Promise.all([
      this.sendOrderConfirmation(order),
      this.updateInventory(order.line_items),
      this.createShippingLabel(order),
      this.syncToERP(order)
    ])
  }
 
  validateOrderData(order) {
    const requiredFields = ['id', 'order_number', 'email', 'line_items']
    
    for (const field of requiredFields) {
      if (!order[field]) {
        throw new Error(`Missing required field: ${field}`)
      }
    }
 
    if (!Array.isArray(order.line_items) || order.line_items.length === 0) {
      throw new Error('Order must have at least one line item')
    }
  }
}
 
// 使用示例
const processor = new WebhookProcessor()
 
app.post('/webhooks/:topic', async (req, res) => {
  try {
    const topic = req.params.topic.replace('-', '/')
    const data = JSON.parse(req.body)
    
    // 立即返回成功响应
    res.status(200).send('Webhook received')
    
    // 异步处理webhook
    setImmediate(() => {
      processor.processWebhook(topic, data)
    })
  } catch (error) {
    console.error('Error parsing webhook data:', error)
    res.status(400).send('Invalid webhook data')
  }
})

// 使用Redis队列处理Webhook
const Queue = require('bull')
const Redis = require('redis')
 
const redisClient = Redis.createClient(process.env.REDIS_URL)
const webhookQueue = new Queue('webhook processing', {
  redis: {
    port: process.env.REDIS_PORT,
    host: process.env.REDIS_HOST,
    password: process.env.REDIS_PASSWORD
  }
})
 
// 队列处理器
webhookQueue.process('order-create', 5, async (job) => {
  const { order } = job.data
  
  try {
    await processOrderCreation(order)
    console.log(`Order ${order.order_number} processed successfully`)
  } catch (error) {
    console.error(`Failed to process order ${order.order_number}:`, error)
    throw error // 这将触发重试
  }
})
 
webhookQueue.process('product-update', 3, async (job) => {
  const { product } = job.data
  
  try {
    await processProductUpdate(product)
    console.log(`Product ${product.handle} updated successfully`)
  } catch (error) {
    console.error(`Failed to update product ${product.handle}:`, error)
    throw error
  }
})
 
// Webhook接收器
app.post('/webhooks/orders/create', async (req, res) => {
  try {
    const order = JSON.parse(req.body)
    
    // 添加到队列
    await webhookQueue.add('order-create', { order }, {
      attempts: 5,
      backoff: {
        type: 'exponential',
        delay: 2000
      },
      removeOnComplete: 10,
      removeOnFail: 5
    })
    
    res.status(200).send('Order webhook queued for processing')
  } catch (error) {
    console.error('Error queueing order webhook:', error)
    res.status(500).send('Error queueing webhook')
  }
})
 
// 监控队列状态
webhookQueue.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed with result:`, result)
})
 
webhookQueue.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed:`, err)
})
 
webhookQueue.on('stalled', (job) => {
  console.warn(`Job ${job.id} stalled`)
})

// 智能路由和条件处理
class SmartWebhookRouter {
  constructor() {
    this.rules = new Map()
    this.filters = new Map()
  }
 
  // 添加处理规则
  addRule(topic, condition, handler) {
    if (!this.rules.has(topic)) {
      this.rules.set(topic, [])
    }
    
    this.rules.get(topic).push({
      condition,
      handler
    })
  }
 
  // 添加过滤器
  addFilter(topic, filter) {
    if (!this.filters.has(topic)) {
      this.filters.set(topic, [])
    }
    
    this.filters.get(topic).push(filter)
  }
 
  async processWebhook(topic, data) {
    // 应用过滤器
    const filters = this.filters.get(topic) || []
    for (const filter of filters) {
      if (!filter(data)) {
        console.log(`Webhook filtered out: ${topic}`)
        return
      }
    }
 
    // 应用规则
    const rules = this.rules.get(topic) || []
    for (const rule of rules) {
      if (rule.condition(data)) {
        await rule.handler(data)
      }
    }
  }
}
 
// 使用示例
const router = new SmartWebhookRouter()
 
// 添加过滤器 - 只处理已支付的订单
router.addFilter('orders/create', (order) => {
  return order.financial_status === 'paid'
})
 
// 添加条件处理规则
router.addRule('orders/create', 
  (order) => order.total_price >= 500, // 条件：大额订单
  async (order) => {
    await notifyVIPCustomerService(order)
    await applyVIPDiscount(order.customer.id)
  }
)
 
router.addRule('orders/create',
  (order) => order.shipping_address.country !== 'US', // 条件：国际订单
  async (order) => {
    await processInternationalShipping(order)
    await calculateCustomsDuties(order)
  }
)
 
router.addRule('orders/create',
  (order) => order.line_items.some(item => item.product_type === 'digital'), // 条件：包含数字产品
  async (order) => {
    await generateDigitalDelivery(order)
    await sendDownloadLinks(order.customer.email, order.line_items)
  }
)

// Webhook聚合处理器
class WebhookAggregator {
  constructor(options = {}) {
    this.batchSize = options.batchSize || 10
    this.batchTimeout = options.batchTimeout || 5000 // 5秒
    this.batches = new Map()
    this.timers = new Map()
  }
 
  addToBatch(topic, data) {
    // 初始化批次
    if (!this.batches.has(topic)) {
      this.batches.set(topic, [])
    }
 
    // 添加数据到批次
    this.batches.get(topic).push(data)
 
    // 检查是否达到批次大小
    if (this.batches.get(topic).length >= this.batchSize) {
      this.processBatch(topic)
    } else {
      // 设置超时处理
      this.resetTimer(topic)
    }
  }
 
  resetTimer(topic) {
    // 清除现有定时器
    if (this.timers.has(topic)) {
      clearTimeout(this.timers.get(topic))
    }
 
    // 设置新的定时器
    const timer = setTimeout(() => {
      this.processBatch(topic)
    }, this.batchTimeout)
 
    this.timers.set(topic, timer)
  }
 
  async processBatch(topic) {
    const batch = this.batches.get(topic) || []
    
    if (batch.length === 0) {
      return
    }
 
    // 清除定时器
    if (this.timers.has(topic)) {
      clearTimeout(this.timers.get(topic))
      this.timers.delete(topic)
    }
 
    // 重置批次
    this.batches.set(topic, [])
 
    try {
      console.log(`Processing batch of ${batch.length} ${topic} webhooks`)
      await this.executeBatchHandler(topic, batch)
    } catch (error) {
      console.error(`Error processing batch for ${topic}:`, error)
      // 可以实现重试逻辑
    }
  }
 
  async executeBatchHandler(topic, batch) {
    switch (topic) {
      case 'products/update':
        await this.batchUpdateProductSearch(batch)
        break
      case 'orders/create':
        await this.batchProcessOrders(batch)
        break
      case 'customers/create':
        await this.batchSyncCustomers(batch)
        break
      default:
        console.warn(`No batch handler for topic: ${topic}`)
    }
  }
 
  async batchUpdateProductSearch(products) {
    // 批量更新搜索索引
    const searchUpdates = products.map(product => ({
      id: product.id,
      title: product.title,
      description: product.body_html,
      price: product.variants[0]?.price,
      tags: product.tags
    }))
 
    await updateSearchIndexBatch(searchUpdates)
  }
 
  async batchProcessOrders(orders) {
    // 批量处理订单
    const notifications = []
    const inventoryUpdates = []
 
    for (const order of orders) {
      notifications.push({
        email: order.email,
        orderNumber: order.order_number,
        total: order.total_price
      })
 
      for (const item of order.line_items) {
        inventoryUpdates.push({
          variantId: item.variant_id,
          quantity: -item.quantity
        })
      }
    }
 
    await Promise.all([
      sendBatchNotifications(notifications),
      updateInventoryBatch(inventoryUpdates)
    ])
  }
}
 
// 使用聚合器
const aggregator = new WebhookAggregator({
  batchSize: 20,
  batchTimeout: 3000
})
 
app.post('/webhooks/products/update', (req, res) => {
  const product = JSON.parse(req.body)
  aggregator.addToBatch('products/update', product)
  res.status(200).send('OK')
})

// Webhook监控和分析
class WebhookMonitor {
  constructor() {
    this.metrics = {
      received: 0,
      processed: 0,
      failed: 0,
      averageProcessingTime: 0,
      topicStats: new Map()
    }
    this.startTime = Date.now()
  }
 
  recordWebhookReceived(topic) {
    this.metrics.received++
    this.updateTopicStats(topic, 'received')
  }
 
  recordWebhookProcessed(topic, processingTime) {
    this.metrics.processed++
    this.updateTopicStats(topic, 'processed')
    this.updateAverageProcessingTime(processingTime)
  }
 
  recordWebhookFailed(topic, error) {
    this.metrics.failed++
    this.updateTopicStats(topic, 'failed')
    this.logError(topic, error)
  }
 
  updateTopicStats(topic, type) {
    if (!this.metrics.topicStats.has(topic)) {
      this.metrics.topicStats.set(topic, {
        received: 0,
        processed: 0,
        failed: 0
      })
    }
 
    this.metrics.topicStats.get(topic)[type]++
  }
 
  updateAverageProcessingTime(processingTime) {
    const current = this.metrics.averageProcessingTime
    const processed = this.metrics.processed
    
    this.metrics.averageProcessingTime = 
      (current * (processed - 1) + processingTime) / processed
  }
 
  logError(topic, error) {
    console.error(`Webhook error for ${topic}:`, {
      message: error.message,
      stack: error.stack,
      timestamp: new Date().toISOString()
    })
  }
 
  getStats() {
    const uptime = Date.now() - this.startTime
    
    return {
      uptime: uptime,
      metrics: this.metrics,
      healthStatus: this.getHealthStatus()
    }
  }
 
  getHealthStatus() {
    const failureRate = this.metrics.failed / this.metrics.received
    
    if (failureRate > 0.1) {
      return 'unhealthy'
    } else if (failureRate > 0.05) {
      return 'warning'
    } else {
      return 'healthy'
    }
  }
 
  // 导出监控数据到外部系统
  async exportMetrics() {
    const stats = this.getStats()
    
    // 发送到监控服务 (如Prometheus, DataDog等)
    await this.sendToMonitoringService(stats)
  }
 
  async sendToMonitoringService(stats) {
    // 实现发送逻辑
    console.log('Webhook metrics:', stats)
  }
}
 
// 集成监控
const monitor = new WebhookMonitor()
 
// 监控中间件
function monitoringMiddleware(req, res, next) {
  const startTime = Date.now()
  const topic = req.get('X-Shopify-Topic')
  
  monitor.recordWebhookReceived(topic)
  
  // 监控响应
  const originalSend = res.send
  res.send = function(data) {
    const processingTime = Date.now() - startTime
    
    if (res.statusCode >= 200 && res.statusCode < 300) {
      monitor.recordWebhookProcessed(topic, processingTime)
    } else {
      monitor.recordWebhookFailed(topic, new Error(`HTTP ${res.statusCode}`))
    }
    
    originalSend.call(this, data)
  }
  
  next()
}
 
// 应用监控中间件
app.use('/webhooks', monitoringMiddleware)
 
// 监控端点
app.get('/webhook-stats', (req, res) => {
  res.json(monitor.getStats())
})
 
// 定期导出指标
setInterval(() => {
  monitor.exportMetrics()
}, 60000) // 每分钟导出一次

// Webhook调试工具
class WebhookDebugger {
  constructor() {
    this.requestLog = []
    this.maxLogSize = 1000
  }
 
  logRequest(req, res) {
    const logEntry = {
      timestamp: new Date().toISOString(),
      headers: req.headers,
      body: req.body,
      query: req.query,
      params: req.params,
      statusCode: res.statusCode,
      processingTime: res.locals.processingTime
    }
 
    this.requestLog.unshift(logEntry)
    
    // 保持日志大小限制
    if (this.requestLog.length > this.maxLogSize) {
      this.requestLog = this.requestLog.slice(0, this.maxLogSize)
    }
  }
 
  getRecentRequests(limit = 50) {
    return this.requestLog.slice(0, limit)
  }
 
  searchRequests(criteria) {
    return this.requestLog.filter(entry => {
      if (criteria.topic && !entry.headers['x-shopify-topic']?.includes(criteria.topic)) {
        return false
      }
      
      if (criteria.shop && !entry.headers['x-shopify-shop-domain']?.includes(criteria.shop)) {
        return false
      }
      
      if (criteria.status && entry.statusCode !== criteria.status) {
        return false
      }
      
      if (criteria.from && new Date(entry.timestamp) < new Date(criteria.from)) {
        return false
      }
      
      if (criteria.to && new Date(entry.timestamp) > new Date(criteria.to)) {
        return false
      }
      
      return true
    })
  }
 
  validateWebhookData(topic, data) {
    const validators = {
      'orders/create': this.validateOrderData,
      'products/update': this.validateProductData,
      'customers/create': this.validateCustomerData
    }
 
    const validator = validators[topic]
    if (validator) {
      return validator(data)
    }
 
    return { valid: true, errors: [] }
  }
 
  validateOrderData(order) {
    const errors = []
    
    if (!order.id) errors.push('Missing order ID')
    if (!order.order_number) errors.push('Missing order number')
    if (!order.email) errors.push('Missing customer email')
    if (!order.line_items || !Array.isArray(order.line_items)) {
      errors.push('Invalid or missing line items')
    }
    
    return {
      valid: errors.length === 0,
      errors
    }
  }
 
  validateProductData(product) {
    const errors = []
    
    if (!product.id) errors.push('Missing product ID')
    if (!product.title) errors.push('Missing product title')
    if (!product.handle) errors.push('Missing product handle')
    
    return {
      valid: errors.length === 0,
      errors
    }
  }
 
  validateCustomerData(customer) {
    const errors = []
    
    if (!customer.id) errors.push('Missing customer ID')
    if (!customer.email) errors.push('Missing customer email')
    
    return {
      valid: errors.length === 0,
      errors
    }
  }
}
 
// 调试中间件
const debugger = new WebhookDebugger()
 
function debugMiddleware(req, res, next) {
  const startTime = Date.now()
  
  const originalSend = res.send
  res.send = function(data) {
    res.locals.processingTime = Date.now() - startTime
    debugger.logRequest(req, res)
    originalSend.call(this, data)
  }
  
  next()
}
 
// 调试端点
app.get('/webhook-debug/recent', (req, res) => {
  const limit = parseInt(req.query.limit) || 50
  res.json(debugger.getRecentRequests(limit))
})
 
app.get('/webhook-debug/search', (req, res) => {
  const results = debugger.searchRequests(req.query)
  res.json(results)
})
 
app.post('/webhook-debug/validate', (req, res) => {
  const { topic, data } = req.body
  const validation = debugger.validateWebhookData(topic, data)
  res.json(validation)
})

// 高性能异步处理
class HighPerformanceWebhookProcessor {
  constructor(options = {}) {
    this.concurrency = options.concurrency || 10
    this.activeJobs = 0
    this.jobQueue = []
    this.circuitBreaker = new CircuitBreaker()
  }
 
  async processWebhook(topic, data) {
    return new Promise((resolve, reject) => {
      const job = { topic, data, resolve, reject }
      
      if (this.activeJobs < this.concurrency) {
        this.executeJob(job)
      } else {
        this.jobQueue.push(job)
      }
    })
  }
 
  async executeJob(job) {
    this.activeJobs++
    
    try {
      // 使用断路器保护
      await this.circuitBreaker.execute(async () => {
        await this.handleWebhook(job.topic, job.data)
      })
      
      job.resolve()
    } catch (error) {
      job.reject(error)
    } finally {
      this.activeJobs--
      this.processQueue()
    }
  }
 
  processQueue() {
    if (this.jobQueue.length > 0 && this.activeJobs < this.concurrency) {
      const job = this.jobQueue.shift()
      this.executeJob(job)
    }
  }
 
  async handleWebhook(topic, data) {
    // 实际的Webhook处理逻辑
    const handlers = {
      'orders/create': this.handleOrderCreate.bind(this),
      'orders/updated': this.handleOrderUpdate.bind(this),
      'products/update': this.handleProductUpdate.bind(this)
    }
 
    const handler = handlers[topic]
    if (handler) {
      await handler(data)
    }
  }
}
 
// 断路器实现
class CircuitBreaker {
  constructor(options = {}) {
    this.failureThreshold = options.failureThreshold || 5
    this.timeout = options.timeout || 60000 // 1分钟
    this.monitoringPeriod = options.monitoringPeriod || 10000 // 10秒
    
    this.state = 'CLOSED' // CLOSED, OPEN, HALF_OPEN
    this.failureCount = 0
    this.lastFailureTime = null
  }
 
  async execute(operation) {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailureTime >= this.timeout) {
        this.state = 'HALF_OPEN'
      } else {
        throw new Error('Circuit breaker is OPEN')
      }
    }
 
    try {
      const result = await operation()
      this.onSuccess()
      return result
    } catch (error) {
      this.onFailure()
      throw error
    }
  }
 
  onSuccess() {
    this.failureCount = 0
    this.state = 'CLOSED'
  }
 
  onFailure() {
    this.failureCount++
    this.lastFailureTime = Date.now()
    
    if (this.failureCount >= this.failureThreshold) {
      this.state = 'OPEN'
    }
  }
}