feat(sync): cn event coverage, bootstrap and outbox sender

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-18 15:21:41 +08:00
parent 3d6e9ce175
commit 65726702f3
43 changed files with 3333 additions and 181 deletions
@@ -0,0 +1,128 @@
+// services/sync/src/main.ts —— sync 发送器进程入口（P2 Task 3）。构建产物
+// dist/main.js 由根 dev:sync / 生产启动命令运行；env 一律来自进程环境（根
+// dotenv-cli 注入），此处不读文件。循环：每 SYNC_PUSH_INTERVAL_MS 轮询一次
+// outbox（SYNC_BATCH_SIZE 一批，high 优先——见 sender.ts 排序说明；满批立即
+// 追下一轮，积压排空不被轮询间隔拖慢）。失败退避：连续失败按间隔 ×2^n 指数
+// 递增、60s 封顶，成功即归零。优雅退出：SIGINT/SIGTERM 停轮询（等待在飞
+// 一轮，fetch 超时已封顶 30s）后断开 Prisma。
+import { pino } from 'pino';
+import { PrismaClient } from '@photo/db-cn';
+import {
+  SYNC_BATCH_SIZE,
+  SYNC_INTERNAL_KEY_ENV,
+  SYNC_PUSH_INTERVAL_MS,
+} from '@photo/contracts';
+import { OutboxSender } from './sender';
+
+/** 失败退避封顶（对齐 SYNC_RECONCILE_INTERVAL_MS 量级：分钟级封顶足够）。 */
+const MAX_BACKOFF_MS = 60_000;
+/** 单次 HTTP 请求超时。 */
+const REQUEST_TIMEOUT_MS = 30_000;
+
+/** 读正整数 env（可覆盖契约默认值）；非法值回退默认并告警。 */
+function readIntEnv(
+  logger: { warn: (obj: object, msg: string) => void },
+  name: string,
+  fallback: number,
+): number {
+  const raw = process.env[name];
+  if (raw === undefined || raw === '') {
+    return fallback;
+  }
+  const parsed = Number.parseInt(raw, 10);
+  if (!Number.isFinite(parsed) || parsed <= 0) {
+    logger.warn({ name, raw }, 'env 非正整数，回退契约默认值');
+    return fallback;
+  }
+  return parsed;
+}
+
+const sleep = (ms: number): Promise<void> =>
+  new Promise((resolve) => setTimeout(resolve, ms));
+
+async function main(): Promise<void> {
+  const logger = pino({ level: process.env.LOG_LEVEL ?? 'info' });
+  const intervalMs = readIntEnv(logger, 'SYNC_PUSH_INTERVAL_MS', SYNC_PUSH_INTERVAL_MS);
+  const batchSize = readIntEnv(logger, 'SYNC_BATCH_SIZE', SYNC_BATCH_SIZE);
+  const ovApiUrl = process.env.OV_API_URL ?? 'http://127.0.0.1:8081';
+  const internalKey = process.env[SYNC_INTERNAL_KEY_ENV] ?? '';
+  if (internalKey === '') {
+    // 与接收端 guard 同一取舍：密钥缺失宁可不可用，绝不裸推。
+    logger.error({ env: SYNC_INTERNAL_KEY_ENV }, '缺少内部同步密钥，发送器退出');
+    process.exit(1);
+  }
+
+  const prisma = new PrismaClient();
+  const sender = new OutboxSender({
+    prisma,
+    ovApiUrl,
+    internalKey,
+    batchSize,
+    timeoutMs: REQUEST_TIMEOUT_MS,
+    logger,
+  });
+  logger.info(
+    { ovApiUrl, intervalMs, batchSize },
+    'sync sender started',
+  );
+
+  let stopping = false;
+  let inFlight: Promise<void> | null = null;
+  const loop = async (): Promise<void> => {
+    let failureStreak = 0;
+    while (!stopping) {
+      try {
+        const stats = await sender.runOnce();
+        failureStreak = 0;
+        if (stats.picked > 0) {
+          logger.info(stats, 'outbox batch pushed');
+        }
+        if (stats.picked >= batchSize) {
+          continue; // 满批：积压未清，立即追下一轮
+        }
+      } catch (err) {
+        failureStreak += 1;
+        const backoffMs = Math.min(
+          intervalMs * 2 ** (failureStreak - 1),
+          MAX_BACKOFF_MS,
+        );
+        logger.error(
+          {
+            err: err instanceof Error ? err.message : String(err),
+            failureStreak,
+            backoffMs,
+          },
+          'outbox push failed（整批未标记，下轮重推）',
+        );
+        await sleep(backoffMs);
+        continue;
+      }
+      await sleep(intervalMs);
+    }
+  };
+  inFlight = loop();
+
+  const shutdown = (signal: string): void => {
+    if (stopping) return;
+    stopping = true;
+    logger.info({ signal }, 'sync sender shutting down');
+    void (async () => {
+      try {
+        // 等在飞一轮收尾（runOnce 各步均有界：HTTP 30s 超时 / DB 查询即回）。
+        await inFlight;
+        await prisma.$disconnect();
+      } catch (err) {
+        logger.error(
+          { err: err instanceof Error ? err.message : String(err) },
+          'sync sender 关闭异常',
+        );
+      } finally {
+        process.exit(0);
+      }
+    })();
+  };
+  process.once('SIGINT', () => shutdown('SIGINT'));
+  process.once('SIGTERM', () => shutdown('SIGTERM'));
+}
+
+void main();
@@ -0,0 +1,251 @@
+// services/sync/src/sender.ts —— cn→ov outbox 发送器（P2 Task 3）。
+// 单次推送 runOnce()：从 cn 库 outbox_events 取 deliveredAt 为空的事件一批，
+// 转内部同步线格式（openapi SyncEvent）后 HMAC 签名推送 ov 入口
+// POST /v1/internal/sync/content-batches；响应 acceptedEventIds 即「受理」锚点
+// （含幂等重复），据此回写 deliveredAt。
+//
+// 高优先级排序：**显式两段取数**（先 high 后其余，段内 occurredAt 升序）而非
+// ORDER BY priority DESC——'high' < 'normal'（字典序 h<n），DESC 反而把 normal
+// 排前；两段取数不依赖任何字符串序假设，新增优先级值也不失效。间隔轮询模型下
+// high 事件最坏等待一个 SYNC_PUSH_INTERVAL_MS（默认 2s，D11「high 不受间隔
+// 约束」的落地取舍：2s 上限可接受，专路即时唤醒属 YAGNI）。
+//
+// 失败语义：任一环节抛错则本 tick 整批不标记、下轮重推（eventId 幂等保证重试
+// 安全）；退避由 main 循环负责（指数递增封顶），本模块无内部重试。
+import type { PrismaClient } from '@photo/db-cn';
+import type { components } from '@photo/contracts';
+import {
+  type OutboxEvent,
+  SYNC_SIGNATURE_HEADER,
+  SYNC_TIMESTAMP_HEADER,
+  isTombstoneOutboxType,
+  signSyncBody,
+  syncEntityTypeOfOutboxType,
+} from '@photo/contracts';
+import type { Logger } from 'pino';
+
+/** openapi SyncEvent（线格式；payload 为实体类型相关载荷对象）。 */
+export type SyncEventWire = components['schemas']['SyncEvent'];
+
+/** ov content-batches 响应（openapi ContentBatchResponse）。 */
+interface ContentBatchResponse {
+  acceptedEventIds: string[];
+  applied: number;
+}
+
+/** outbox 行（发送器只读关心的列）。 */
+type OutboxRow = {
+  id: string;
+  type: string;
+  albumId: string;
+  photoId: string | null;
+  entityVersion: number;
+  priority: string;
+  payload: unknown;
+  occurredAt: Date;
+};
+
+export interface OutboxSenderOptions {
+  prisma: PrismaClient;
+  /** ov 入口基址（如 http://127.0.0.1:8081，不含路径）；端点路径本模块拼接。 */
+  ovApiUrl: string;
+  /** HMAC 密钥（与接收端 InternalSyncGuard 同一 SYNC_INTERNAL_KEY 值）。 */
+  internalKey: string;
+  /** 单批上限（默认契约 SYNC_BATCH_SIZE）。 */
+  batchSize?: number;
+  /** 单次 HTTP 超时毫秒（默认 30s）。 */
+  timeoutMs?: number;
+  logger?: Logger;
+}
+
+/** 单次 runOnce 的观测口径（main 循环打点；测试用于排空/排序判断）。 */
+export interface PushStats {
+  /** 本批取到的事件数（0 则跳过 HTTP）。 */
+  picked: number;
+  /** ov 受理的事件 ID 清单（含幂等重复；批内顺序即受理顺序）。 */
+  acceptedEventIds: string[];
+  /** ov 实际应用（新版本生效）的条数。 */
+  applied: number;
+}
+
+const DEFAULT_BATCH_SIZE = 100;
+const DEFAULT_TIMEOUT_MS = 30_000;
+
+/** ov 侧 content-batches 端点路径（全局前缀 v1 + InternalSyncGuard 域）。 */
+const CONTENT_BATCHES_PATH = '/v1/internal/sync/content-batches';
+
+export class OutboxSender {
+  private readonly batchSize: number;
+  private readonly timeoutMs: number;
+  private readonly endpoint: string;
+
+  constructor(private readonly options: OutboxSenderOptions) {
+    this.batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE;
+    this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+    // 基址容错：吞并尾斜杠再拼路径，端点只此一处。
+    this.endpoint = `${options.ovApiUrl.replace(/\/+$/, '')}${CONTENT_BATCHES_PATH}`;
+  }
+
+  /** 单轮推送：取批 → 线格式 → 签名推送 → 按受理回写 deliveredAt。 */
+  async runOnce(): Promise<PushStats> {
+    const rows = await this.pickBatch();
+    if (rows.length === 0) {
+      return { picked: 0, acceptedEventIds: [], applied: 0 };
+    }
+    const events = rows.map((row) => this.toWireEvent(row));
+    const response = await this.push(events);
+    // 只标记 ov 明确受理的 eventId（且仍为未投递），幂等可重入。
+    if (response.acceptedEventIds.length > 0) {
+      await this.options.prisma.outbox_events.updateMany({
+        where: {
+          id: { in: response.acceptedEventIds },
+          deliveredAt: null,
+        },
+        data: { deliveredAt: new Date() },
+      });
+    }
+    this.options.logger?.debug(
+      {
+        picked: rows.length,
+        accepted: response.acceptedEventIds.length,
+        applied: response.applied,
+      },
+      'outbox batch pushed',
+    );
+    return {
+      picked: rows.length,
+      acceptedEventIds: response.acceptedEventIds,
+      applied: response.applied,
+    };
+  }
+
+  /**
+   * 两段取数（见文件头排序说明）：high 段占满批则不发第二查；两查之间存在
+   * 微小时序窗口（新事件落入两查之间）——对轮询器无害，下一轮补上。
+   */
+  private async pickBatch(): Promise<OutboxRow[]> {
+    const prisma = this.options.prisma;
+    const high = await prisma.outbox_events.findMany({
+      where: { deliveredAt: null, priority: 'high' },
+      orderBy: [{ occurredAt: 'asc' }, { id: 'asc' }],
+      take: this.batchSize,
+      // 只取发送器关心的列（payload JsonValue 原样透传）。
+      select: {
+        id: true,
+        type: true,
+        albumId: true,
+        photoId: true,
+        entityVersion: true,
+        priority: true,
+        payload: true,
+        occurredAt: true,
+      },
+    });
+    if (high.length >= this.batchSize) {
+      return high;
+    }
+    const rest = await prisma.outbox_events.findMany({
+      where: { deliveredAt: null, priority: { not: 'high' } },
+      orderBy: [{ occurredAt: 'asc' }, { id: 'asc' }],
+      take: this.batchSize - high.length,
+      select: {
+        id: true,
+        type: true,
+        albumId: true,
+        photoId: true,
+        entityVersion: true,
+        priority: true,
+        payload: true,
+        occurredAt: true,
+      },
+    });
+    return [...high, ...rest];
+  }
+
+  /**
+   * outbox 行 → SyncEvent 线格式。entityId 按实体类型取自不同位置：
+   * album → albumId 列；photo → photoId 列；category/asset → 载荷内的
+   * categoryId/assetId（outbox 表无此二列）。映射缺料抛错（fail-fast：
+   * 本轮不推、不标记，错误日志暴露；cn 侧发射器按构造保证不缺料）。
+   */
+  private toWireEvent(row: OutboxRow): SyncEventWire {
+    const type = row.type as OutboxEvent['type'];
+    const entityType = syncEntityTypeOfOutboxType(type);
+    const payload = (row.payload ?? {}) as Record<string, unknown>;
+    let entityId: string;
+    if (entityType === 'album') {
+      entityId = row.albumId;
+    } else if (entityType === 'photo') {
+      if (row.photoId === null) {
+        throw new Error(`outbox 事件 ${row.id}（${type}）缺 photoId，无法推导 entityId`);
+      }
+      entityId = row.photoId;
+    } else if (typeof payload.categoryId === 'string') {
+      entityId = payload.categoryId;
+    } else if (typeof payload.assetId === 'string') {
+      entityId = payload.assetId;
+    } else {
+      throw new Error(`outbox 事件 ${row.id}（${type}）载荷缺实体 ID，无法推导 entityId`);
+    }
+    const tombstone = isTombstoneOutboxType(type);
+    return {
+      eventId: row.id,
+      schemaVersion: 1,
+      entityType,
+      entityId,
+      albumId: row.albumId,
+      entityVersion: row.entityVersion,
+      // 线格式 tombstone 显式携带（生成类型必填：openapi-typescript 把带
+      // default 的属性渲染为非可选）。
+      tombstone,
+      priority: row.priority === 'high' ? 'high' : 'normal',
+      payload: payload as SyncEventWire['payload'],
+    };
+  }
+
+  /**
+   * 签名推送：同一 JSON.stringify 产物既作签名输入也作请求体（原始字节一致，
+   * 接收端 rawBody 验签才能通过——见 contracts/internal-sync.ts 头注）。
+   * 非 2xx / 响应形状异常均抛错（调用方整批重试）。
+   */
+  private async push(events: SyncEventWire[]): Promise<ContentBatchResponse> {
+    const rawBody = JSON.stringify({ batch: events });
+    const timestamp = Math.floor(Date.now() / 1000).toString();
+    let response: Response;
+    try {
+      response = await fetch(this.endpoint, {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+          [SYNC_TIMESTAMP_HEADER]: timestamp,
+          [SYNC_SIGNATURE_HEADER]: signSyncBody(this.options.internalKey, rawBody),
+        },
+        body: rawBody,
+        signal: AbortSignal.timeout(this.timeoutMs),
+      });
+    } catch (err) {
+      throw new Error(
+        `content-batches 请求失败（${this.endpoint}）：${err instanceof Error ? err.message : String(err)}`,
+      );
+    }
+    if (!response.ok) {
+      const body = await response.text().catch(() => '');
+      throw new Error(
+        `content-batches 非 2xx（${this.endpoint}）：status=${response.status} body=${body.slice(0, 500)}`,
+      );
+    }
+    const data: unknown = await response.json();
+    if (
+      typeof data !== 'object' ||
+      data === null ||
+      !Array.isArray((data as ContentBatchResponse).acceptedEventIds) ||
+      !((data as ContentBatchResponse).acceptedEventIds as unknown[]).every(
+        (id) => typeof id === 'string',
+      ) ||
+      typeof (data as ContentBatchResponse).applied !== 'number'
+    ) {
+      throw new Error('content-batches 响应形状异常（acceptedEventIds/applied 缺失）');
+    }
+    return data as ContentBatchResponse;
+  }
+}