GraphQL 事件驱动集成：订阅、Webhook 与消息队列

# 事件类型定义（可作为 GraphQL 类型暴露，也可映射到 Avro/Protobuf）
type OrderCreatedEvent {
  eventId: ID!            # 全局唯一，用于幂等
  occurredAt: DateTime!   # 事件发生时间（不是投递时间）
  aggregateId: ID!        # 聚合根 ID（订单 ID）
  version: Int!           # 事件版本，用于演进
  payload: OrderCreatedPayload!
}

type OrderCreatedPayload {
  orderId: ID!
  buyerId: ID!
  total: Money!
  items: [OrderItemSnapshot!]!
}

// 统一信封：元数据与业务载荷分离
interface EventEnvelope<T> {
  eventId: string;
  eventType: string;        // "order.created"
  eventVersion: number;
  occurredAt: string;       // ISO 8601
  traceId: string;          // 全链路追踪
  producer: string;         // 生产服务名
  payload: T;
}

# 事件类型命名：<domain>.<aggregate>.<action>
order.created       # 订单已创建
order.paid          # 订单已支付
payment.refunded    # 支付已退款
inventory.reserved  # 库存已预留

# ❌ 避免命令式命名
create_order        # 这是命令，不是事件

# 客户端订阅：只关心自己订单的状态变化
subscription OnOrderStatus($orderId: ID!) {
  orderStatusChanged(orderId: $orderId) {
    orderId
    status
    updatedAt
  }
}

import { PubSub } from 'graphql-subscriptions';

const pubsub = new PubSub();

const resolvers = {
  Subscription: {
    orderStatusChanged: {
      subscribe: (_p, { orderId }, ctx) => {
        // 鉴权：确认用户有权订阅该订单
        assertCanViewOrder(ctx.user, orderId);
        // 过滤：只推送该订单的事件
        return pubsub.asyncIterator(`order.status.${orderId}`);
      },
    },
  },
  Mutation: {
    updateOrderStatus: async (_p, { id, status }, ctx) => {
      const order = await orderService.update(ctx, id, status);
      // 发布事件到对应频道
      await pubsub.publish(`order.status.${id}`, {
        orderStatusChanged: { orderId: id, status, updatedAt: new Date() },
      });
      return order;
    },
  },
};

// 多实例：用 Redis PubSub 替换内存实现
import { RedisPubSub } from 'graphql-redis-subscriptions';
const pubsub = new RedisPubSub({
  publisher: new Redis(process.env.REDIS_URL),
  subscriber: new Redis(process.env.REDIS_URL),
});

import { createHmac } from 'node:crypto';

// 发送方：对 (timestamp + body) 签名
function sign(body: string, timestamp: number, secret: string): string {
  return createHmac('sha256', secret)
    .update(`${timestamp}.${body}`)
    .digest('hex');
}

// 投递时附带签名头
await fetch(endpoint, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'X-Webhook-Id': event.eventId,
    'X-Webhook-Timestamp': String(Date.now()),
    'X-Webhook-Signature': sign(JSON.stringify(event), Date.now(), secret),
  },
  body: JSON.stringify(event),
});

// 接收方：验签 + 防重放（时间窗口）
function verify(raw: string, sig: string, ts: string, secret: string): boolean {
  if (Date.now() - Number(ts) > 5 * 60 * 1000) return false;  // 超过 5 分钟拒绝
  return timingSafeEqual(Buffer.from(sig), Buffer.from(sign(raw, Number(ts), secret)));
}

// 事务性发件箱（Transactional Outbox）：避免"写库成功但发消息失败"
async function createOrder(tx: Tx, input: CreateOrderInput) {
  const order = await tx.order.create({ data: input });
  // 事件与业务写入同一事务，保证原子性
  await tx.outbox.create({
    data: {
      eventId: randomUUID(),
      eventType: 'order.created',
      aggregateId: order.id,
      payload: JSON.stringify(toEventPayload(order)),
      status: 'PENDING',
    },
  });
  return order;
}

// 指数退避 + 抖动，避免重试风暴
function backoff(attempt: number): number {
  const base = Math.min(2 ** attempt * 1000, 5 * 60 * 1000);  // 上限 5 分钟
  const jitter = Math.random() * 1000;
  return base + jitter;
}

async function deliverWithRetry(job: DeliveryJob) {
  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    try {
      const res = await postWithTimeout(job.endpoint, job.body, 10_000);
      if (res.ok) return { ok: true };
      if (res.status >= 400 && res.status < 500 && res.status !== 429) {
        return { ok: false, permanent: true };   // 4xx（除 429）不重试
      }
    } catch (err) {
      logger.warn({ err, attempt }, 'delivery failed');
    }
    await sleep(backoff(attempt));
  }
  return { ok: false, permanent: false };   // 转入 DLQ
}

// 消费端幂等：用 eventId 去重表
async function handleEvent(event: EventEnvelope<unknown>) {
  const inserted = await prisma.processedEvent.createMany({
    data: [{ eventId: event.eventId }],
    skipDuplicates: true,          // 唯一约束冲突则跳过
  });
  if (inserted.count === 0) {
    logger.info({ eventId: event.eventId }, 'duplicate event ignored');
    return;                        // 已处理过，直接返回
  }
  await applyEvent(event);
}

// 状态机幂等：只在合法状态转移时应用
const VALID: Record<string, string[]> = {
  CREATED: ['PAID', 'CANCELLED'],
  PAID: ['SHIPPED', 'REFUNDED'],
  SHIPPED: ['DELIVERED'],
};
function canTransition(from: string, to: string): boolean {
  return VALID[from]?.includes(to) ?? false;
}

// 版本检测：拒绝乱序的旧事件
if (event.version <= current.version) {
  logger.warn({ eventId: event.eventId }, 'out-of-order event dropped');
  return;
}

// 投递失败 → 写入 DLQ（含失败原因与完整上下文）
async function moveToDlq(job: DeliveryJob, reason: string) {
  await prisma.deadLetter.create({
    data: {
      eventId: job.eventId,
      endpoint: job.endpoint,
      payload: job.body,
      attempts: job.attempts,
      lastError: reason,
      failedAt: new Date(),
    },
  });
  metrics.dlqSize.inc();
  alerts.warn(`Event ${job.eventId} moved to DLQ`);
}

// 用 OpenTelemetry 把事件投递纳入全链路追踪
const span = tracer.startSpan('webhook.deliver', {
  attributes: { 'event.id': event.eventId, 'delivery.attempt': job.attempts },
});