用 Feature Flag 做架构演进的安全阀

// 直接切换数据源，出问题只能回滚代码
async function getOrder(orderId: string) {
  // 改了一行配置就切了，没有任何缓冲
  return await tiDbClient.query('SELECT * FROM orders WHERE id = ?', [orderId])
}

type MigrationStage = 'off' | 'dualwrite' | 'shadow' | 'live' | 'rampdown' | 'complete'
 
async function getOrder(orderId: string) {
  const stage = await getFlag('order-db-migration', 'off') as MigrationStage
 
  switch (stage) {
    case 'off':
      return readFromMySQL(orderId)
 
    case 'dualwrite':
      // 双写阶段还在读老库
      return readFromMySQL(orderId)
 
    case 'shadow': {
      // 新老库同时读，比对结果但只返回老库数据
      const [oldResult, newResult] = await Promise.all([
        readFromMySQL(orderId),
        readFromTiDB(orderId),
      ])
      compareAndAlert(oldResult, newResult, orderId) // 不一致时告警
      return oldResult
    }
 
    case 'live':
    case 'rampdown':
      return readFromTiDB(orderId)
 
    case 'complete':
      return readFromTiDB(orderId)
 
    default:
      return readFromMySQL(orderId) // 兜底走老路
  }
}

async function createOrder(order: OrderInput) {
  const stage = await getFlag('order-db-migration', 'off') as MigrationStage
 
  if (stage === 'off') {
    return writeToMySQL(order)
  }
 
  if (stage === 'dualwrite' || stage === 'shadow') {
    // 先写老库成功，再异步写新库
    const result = await writeToMySQL(order)
    // 新库写入失败不阻塞主流程，但要告警
    writeToTiDB(order).catch(err => {
      logger.error('TiDB dual-write failed', { orderId: order.id, error: err })
      metrics.increment('migration.dualwrite.failure')
    })
    return result
  }
 
  // live 之后只写新库
  return writeToTiDB(order)
}

// 路由级别灰度配置
interface RouteGrayConfig {
  enabled: boolean          // 是否走 Envoy
  weight: number            // 流量百分比 0-100
  fallbackToNginx: boolean  // Envoy 出错是否回退
}
 
// Flag payload 示例
{
  "user-service": { "enabled": true, "weight": 10, "fallbackToNginx": true },
  "order-service": { "enabled": false, "weight": 0, "fallbackToNginx": true },
  "payment-service": { "enabled": true, "weight": 100, "fallbackToNginx": false }
}

# 一刀切，所有流量都走 Envoy
upstream backend {
    server envoy-gateway:10000;
    # nginx 配置全删了，回不去
}

async function routeRequest(req: IncomingRequest) {
  const routeName = extractRouteName(req)
  const config = await getFlag('gateway-migration-routes', {}) as Record<string, RouteGrayConfig>
  const routeConfig = config[routeName] ?? { enabled: false, weight: 0, fallbackToNginx: true }
 
  // 未开启或权重为 0，走老网关
  if (!routeConfig.enabled || routeConfig.weight === 0) {
    return forwardToNginx(req)
  }
 
  // 按权重决定是否走新网关
  if (Math.random() * 100 > routeConfig.weight) {
    return forwardToNginx(req)
  }
 
  try {
    const response = await forwardToEnvoy(req, { timeout: 3000 })
 
    // 监控响应质量，异常时自动降级
    if (response.status >= 500 || response.latency > 5000) {
      metrics.increment('gateway.envoy.degraded', { route: routeName })
      if (routeConfig.fallbackToNginx) {
        return forwardToNginx(req) // 自动回退
      }
    }
 
    return response
  } catch (err) {
    if (routeConfig.fallbackToNginx) {
      metrics.increment('gateway.envoy.fallback', { route: routeName })
      return forwardToNginx(req)
    }
    throw err
  }
}

// Flag 控制消费策略
type ConsumeStrategy = 'old-only' | 'dual-consume' | 'new-only'
 
async function processMessage(rawMessage: RawMessage) {
  const strategy = await getFlag('mq-migration-strategy', 'old-only') as ConsumeStrategy
 
  if (strategy === 'old-only') {
    return processViaRabbitMQ(rawMessage)
  }
 
  if (strategy === 'dual-consume') {
    // 两条队列都消费，但只以老队列结果为准
    const [oldResult, newResult] = await Promise.allSettled([
      processViaRabbitMQ(rawMessage),
      processViaKafka(rawMessage),
    ])
 
    // 比对两个结果，不一致时记录告警
    if (oldResult.status === 'fulfilled' && newResult.status === 'fulfilled') {
      if (!shallowEqual(oldResult.value, newResult.value)) {
        logger.warn('MQ migration result mismatch', {
          messageId: rawMessage.id,
          oldResult: oldResult.value,
          newResult: newResult.value,
        })
        metrics.increment('mq.migration.mismatch')
      }
    }
 
    // 始终返回老队列的处理结果
    if (oldResult.status === 'fulfilled') return oldResult.value
    throw oldResult.reason
  }
 
  // new-only 阶段
  return processViaKafka(rawMessage)
}

// RabbitMQ 消费者
rabbitChannel.consume('orders', async (msg) => {
  await handleOrder(msg.content)
})
 
// Kafka 消费者（同时启动，处理相同消息）
await kafkaConsumer.run({
  eachMessage: async ({ message }) => {
    await handleOrder(message.value)  // 同一条消息被处理两次
  },
})

async function handleOrder(messageContent: Buffer) {
  const parsed = JSON.parse(messageContent.toString())
  const { orderId, idempotencyKey } = parsed
 
  // 幂等检查：无论哪个队列来的消息，同一个 key 只处理一次
  const existing = await redis.get(`order:processed:${idempotencyKey}`)
  if (existing) {
    logger.info('Duplicate message skipped', { orderId, idempotencyKey })
    return
  }
 
  // 业务处理
  const result = await processOrder(parsed)
 
  // 记录已处理，TTL 和幂等窗口对齐（比如 24 小时）
  await redis.set(`order:processed:${idempotencyKey}`, '1', 'EX', 86400)
 
  return result
}

async function migrateUserData(userId: string) {
  const useNewService = await getFlag('user-migration', false)
 
  if (useNewService) {
    // 新服务写入后，老服务的数据就过期了
    const result = await newUserService.migrate(userId)
    // 如果这里出问题，老服务的数据已经被标记为「已迁移」
    await oldUserService.markMigrated(userId)
    return result
  }
 
  return oldUserService.getData(userId)
}

async function migrateUserData(userId: string) {
  const useNewService = await getFlag('user-migration', false)
 
  if (useNewService) {
    try {
      // 新服务写入
      const result = await newUserService.migrate(userId)
      // 不立即标记旧数据为「已迁移」，而是记录一个中间状态
      await oldUserService.markMigrating(userId) // 注意：是 migrating，不是 migrated
      return result
    } catch (err) {
      // 新服务失败，旧数据不受影响
      logger.error('New user service migration failed', { userId, error: err })
      // 降级到旧服务
      return oldUserService.getData(userId)
    }
  }
 
  return oldUserService.getData(userId)
}
 
// 回滚时的清理逻辑
async function rollbackUserMigration(userId: string) {
  // 中间状态的数据可以安全回滚
  const status = await oldUserService.getMigrationStatus(userId)
  if (status === 'migrating') {
    await oldUserService.resetMigrationStatus(userId)
    await newUserService.rollback(userId)
  }
}

// 三年前的「临时」Flag，到现在还在
async function getDashboard(userId: string) {
  const v2 = await getFlag('dashboard-v2', false)
  const newLayout = await getFlag('dashboard-new-layout-2023', false)
  const experiment = await getFlag('dashboard-experiment-q3', false)
  const perf = await getFlag('dashboard-perf-improvement', false)
  // 四层嵌套，没人知道哪些还能删
  if (v2) {
    if (newLayout) {
      if (experiment) { /* ... */ }
    }
  }
}

// 每个 Flag 有明确的元信息
interface FlagMetadata {
  owner: string          // 负责人
  createdAt: string      // 创建时间
  expiresAt: string      // 过期时间
  purpose: string        // 用途说明
  cleanupTicket: string  // 清理工单链接
}
 
// 创建 Flag 时就必须填写
const flagMeta: FlagMetadata = {
  owner: 'team-order',
  createdAt: '2026-06-01',
  expiresAt: '2026-08-01', // 两个月后必须清理
  purpose: '订单库 MySQL → TiDB 迁移',
  cleanupTicket: 'JIRA-1234',
}