低复杂度：    只读 + REST API + 同步调用    → 知识库检索、信息查询
中复杂度：    读写 + REST API + 异步调用     → 工单创建、状态更新
高复杂度：    读写 + 多种协议 + 事务保证     → 订单处理、库存扣减

推荐做法：核心流程用 Webhook 事件驱动，非关键数据用定时轮询兜底。

import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """验证 Webhook 的 HMAC-SHA256 签名"""
    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256
    ).hexdigest()

    # 使用时间安全的比较，防止时序攻击
    return hmac.compare_digest(f"sha256={expected}", signature)

# 使用示例
@app.post("/webhook/salesforce")
async def handle_salesforce_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Salesforce-Signature", "")

    if not verify_webhook_signature(body, signature, WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = parse_cloud_event(body)
    await process_event(event)
    return {"status": "ok"}

{
  "specversion": "1.0",
  "type": "com.salesforce.ticket.updated",
  "source": "/crm/salesforce",
  "id": "evt-12345-abcde",
  "time": "2025-06-15T10:30:00Z",
  "data": {
    "ticket_id": "TKT-7890",
    "status": "escalated",
    "customer_id": "CUST-123",
    "assigned_to": "agent-456"
  }
}

数据库 Binlog → CDC (Debezium) → Kafka → AI 服务消费事件

# 乐观锁模式：读取时获取版本号，写入时验证版本
def update_customer(customer_id: str, data: dict, expected_version: int) -> bool:
    """更新客户数据，如果版本不匹配则拒绝"""
    current = db.get_customer(customer_id)
    if current.version != expected_version:
        raise ConflictError(f"Version mismatch: expected {expected_version}, got {current.version}")

    current.update(data)
    current.version += 1
    db.save(current)
    return True

import asyncio
import random

async def exponential_backoff_retry(func, max_retries=3, base_delay=1.0):
    """指数退避重试，带随机抖动"""
    last_error = None
    for attempt in range(max_retries):
        try:
            return await func()
        except (TimeoutError, ConnectionError) as e:
            last_error = e
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)  # 加抖动防"惊群"
            await asyncio.sleep(delay)
            logger.warning(f"Retry {attempt + 1}/{max_retries} after {delay:.1f}s")
    raise last_error

# Saga 补偿事务示例
class TicketSaga:
    async def execute(self, ticket_data: dict) -> str:
        try:
            # 正向操作
            ticket_id = await self.crm.create_ticket(ticket_data)
            stock = await self.erp.check_stock(ticket_data["product_id"])
            if not stock.sufficient:
                await self.compensate(ticket_id)  # 回滚
                raise InsufficientStockError()
            payment_id = await self.payment.charge(ticket_data["amount"])
            return ticket_id
        except Exception as e:
            # 补偿操作
            await self.compensate(ticket_id)
            raise

    async def compensate(self, ticket_id: str):
        """按正向操作的逆序执行补偿"""
        await self.crm.cancel_ticket(ticket_id)
        # 如果有库存操作，需要恢复库存
        # 如果有支付操作，需要退款

正常请求 → 处理成功 → 返回结果
             ↓ 失败
         重试队列 → 重试成功 → 返回结果
             ↓ 超过最大重试次数
         死信队列 (DLQ) → 人工介入处理

{
  "original_message": {...},
  "error": "Salesforce API returned 500",
  "retry_count": 3,
  "first_attempt": "2025-06-15T10:00:00Z",
  "last_attempt": "2025-06-15T10:07:00Z",
  "context": {"ticket_id": "TKT-7890", "customer_id": "CUST-123"}
}

# API Gateway 限流配置 (Kong)
plugins:
- name: rate-limiting
  config:
    hour: 10000          # 每小时 10000 次请求
    policy: redis        # 用 Redis 做分布式计数
    limit_by: consumer   # 按消费者（而非 IP）限流

- name: request-size-limiting
  config:
    allowed_payload_size: 1  # 最大 1MB 请求体

- name: correlation-id
  config:
    header_name: X-Request-ID  # 全链路追踪 ID

面试官：设计一个 AI + Salesforce 的客户支持系统。

1. 架构（1 分钟）
   → 事件驱动：Salesforce 工单变更 → Webhook → Kafka → AI 服务消费
   → AI 服务用 Temporal 编排：读取工单 → 查客户历史 → 生成回复 → 写回 Salesforce
   → API Gateway 做认证和限流，Sidecar 做 mTLS 加密

2. 数据流（1 分钟）
   → 写操作：AI 通过 Salesforce REST API 更新工单状态（强一致）
   → 读操作：CDC 同步 Salesforce 数据到本地 PostgreSQL（最终一致，用于快速查询）
   → 缓存：Redis 缓存客户画像，TTL 5 分钟

3. 容错设计（1 分钟）
   → Salesforce API 限流（429）：等待 Retry-After 后重试
   → Salesforce API 宕机（5xx）：指数退避重试 3 次，失败后入死信队列
   → AI 服务宕机：Temporal 保证工作流状态持久化，恢复后从断点继续
   → 数据不一致：定期全量比对 + 差异修复任务

4. 安全措施（30 秒）
   → Webhook 签名验证（HMAC-SHA256）
   → Salesforce API 用 OAuth2 Client Credentials
   → 敏感数据（客户信息）加密存储，不落日志
   → API 限流：每 AI 服务实例 1000 req/hour

5. 监控指标（30 秒）
   → 技术指标：API 调用延迟、错误率、队列深度
   → 业务指标：工单自动处理率、平均处理时长、客户满意度