feat(sync): stat snapshot return path with version replacement

P2 Task 9 统计回传（ov→cn）：
- cn 迁移：albums 增 viewCountReplica/statsVersion/statsSyncedAt、
  photos 增 statsVersion（like_count 已有副本位；无 VARCHAR 新列不涉校对手决）
- cn 接收端 POST /v1/internal/sync/stat-snapshots（InternalSyncGuard）：
  整批单事务，逐快照 updateMany WHERE statsVersion < statVersion —— 仅更高
  statVersion 以权威 total 整体替换副本（绝不相加），重复/乱序幂等忽略；
  accepted 只数实际应用条数；views 应用同时推进 statsSyncedAt（A27 展示）
- ov 回传发送器 services/sync/stat-return.ts：聚合 outbox_stat_events 未发送
  行按（kind, photoId|albumId）取 MAX(statVersion) 折叠突发 → HMAC 推送 →
  200 后按计数器标记截至该版本全部源行 sentAt；失败留待下轮重发；main.ts
  第三循环 STAT_RETURN_INTERVAL_MS（默认 60s，env 可调）
- 契约 AlbumSummary + statsSyncedAt（string|null readOnly，+gen）；listAlbums/
  getAlbum 经 toSummaryResponse 暴露（管理 UI 列留 P4a）
- e2e（services/sync stat-return.e2e-spec，7 例）：逐次点赞回传替换式收敛
  （2→5→10 非累加）、浏览窗口去重回传 + statsSyncedAt、重复回传 accepted=0
  值不动（A27 不重复累计）、乱序高版本替换（total 可下调）+ 低版本忽略、
  发送端折叠只发最高版本且被折叠行随批确认、管理列表 likeCount/statsSyncedAt
  展示、新端点 401/400 守卫

Co-Authored-By: Claude Code <noreply@anthropic.com>

services/sync/src/main.ts

@@ -1,21 +1,27 @@
 // services/sync/src/main.ts —— sync 进程入口（P2 Task 3 发送器 + Task 4 对账
-// 调度）。构建产物 dist/main.js 由根 dev:sync / 生产启动命令运行；env 一律来自
-// 进程环境（根 dotenv-cli 注入），此处不读文件。
+// 调度 + Task 9 统计回传）。构建产物 dist/main.js 由根 dev:sync / 生产启动
+// 命令运行；env 一律来自进程环境（根 dotenv-cli 注入），此处不读文件。
 //
-// 两个独立循环：
+// 三个独立循环：
 //   · 发送循环：每 SYNC_PUSH_INTERVAL_MS 轮询 outbox（SYNC_BATCH_SIZE 一批，
 //     high 优先——见 sender.ts 排序说明；满批立即追下一轮，积压排空不被轮询
 //     间隔拖慢）。失败退避：连续失败按间隔 ×2^n 指数递增、60s 封顶，成功归零。
 //   · 对账循环（Task 4）：每 SYNC_RECONCILE_INTERVAL_MS 一轮 SyncReconciler.
 //     runOnce（当前态清扫重放 + digest 比对触发 bootstrap，见 reconciler.ts）。
 //     失败仅记日志、下轮再试（分钟级节奏，无需退避——幂等安全）。
+//   · 统计回传循环（Task 9）：每 STAT_RETURN_INTERVAL_MS 一轮 StatReturnSender.
+//     runReturnOnce（ov outbox_stat_events 聚合 → cn stat-snapshots 版本替换，
+//     见 stat-return.ts）。失败语义同对账：留待下轮重试（cn 侧版本比较保证
+//     幂等），分钟级节奏不另设退避。
 //
-// 双库连接：cn 库（outbox 取数）+ ov 库（副本计数直读）。优雅退出：
-// SIGINT/SIGTERM 停两循环（等待在飞一轮，fetch 超时已封顶 30s）后断开两连接。
+// 双库连接：cn 库（outbox 取数）+ ov 库（副本计数直读/回传队列取数）。
+// 优雅退出：SIGINT/SIGTERM 停全部循环（等待在飞一轮，fetch 超时已封顶 30s）
+// 后断开两连接。
 import { pino } from 'pino';
 import { PrismaClient } from '@photo/db-cn';
 import { PrismaClient as OvPrismaClient } from '@photo/db-ov';
 import {
+  STAT_RETURN_INTERVAL_MS,
   SYNC_BATCH_SIZE,
   SYNC_INTERNAL_KEY_ENV,
   SYNC_PUSH_INTERVAL_MS,
@@ -23,6 +29,7 @@ import {
 } from '@photo/contracts';
 import { OutboxSender } from './sender';
 import { SyncReconciler } from './reconciler';
+import { StatReturnSender } from './stat-return';
 
 /** 失败退避封顶（对齐 SYNC_RECONCILE_INTERVAL_MS 量级：分钟级封顶足够）。 */
 const MAX_BACKOFF_MS = 60_000;
@@ -75,6 +82,11 @@ async function main(): Promise<void> {
     'SYNC_RECONCILE_INTERVAL_MS',
     SYNC_RECONCILE_INTERVAL_MS,
   );
+  const statReturnIntervalMs = readIntEnv(
+    logger,
+    'STAT_RETURN_INTERVAL_MS',
+    STAT_RETURN_INTERVAL_MS,
+  );
   const cnApiUrl = process.env.CN_API_URL ?? 'http://127.0.0.1:8080';
   const ovApiUrl = process.env.OV_API_URL ?? 'http://127.0.0.1:8081';
   const internalKey = process.env[SYNC_INTERNAL_KEY_ENV] ?? '';
@@ -102,9 +114,23 @@ async function main(): Promise<void> {
     timeoutMs: REQUEST_TIMEOUT_MS,
     logger,
   });
+  const statReturn = new StatReturnSender({
+    prismaOv,
+    cnApiUrl,
+    internalKey,
+    timeoutMs: REQUEST_TIMEOUT_MS,
+    logger,
+  });
   logger.info(
-    { cnApiUrl, ovApiUrl, intervalMs, batchSize, reconcileIntervalMs },
-    'sync sender + reconciler started',
+    {
+      cnApiUrl,
+      ovApiUrl,
+      intervalMs,
+      batchSize,
+      reconcileIntervalMs,
+      statReturnIntervalMs,
+    },
+    'sync sender + reconciler + stat-return started',
   );
 
   let stopping = false;
@@ -153,7 +179,27 @@ async function main(): Promise<void> {
       await sleepStoppable(reconcileIntervalMs);
     }
   };
-  const inFlight: Promise<void>[] = [senderLoop(), reconcileLoop()];
+  const statReturnLoop = async (): Promise<void> => {
+    while (!stopping) {
+      try {
+        const stats = await statReturn.runReturnOnce();
+        if (stats.picked > 0) {
+          logger.info(stats, 'stat snapshots returned');
+        }
+      } catch (err) {
+        logger.error(
+          { err: err instanceof Error ? err.message : String(err) },
+          'stat return cycle failed（未确认行留待下轮重发，幂等安全）',
+        );
+      }
+      await sleepStoppable(statReturnIntervalMs);
+    }
+  };
+  const inFlight: Promise<void>[] = [
+    senderLoop(),
+    reconcileLoop(),
+    statReturnLoop(),
+  ];
 
   const shutdown = (signal: string): void => {
     if (stopping) return;

services/sync/src/stat-return.ts

@@ -0,0 +1,241 @@
+// services/sync/src/stat-return.ts —— ov→cn 统计快照回传发送器（P2 Task 9）。
+// 单轮 runReturnOnce()：读 ov 库 outbox_stat_events 未发送行 → 按
+// （kind, photoId|albumId）聚合取 **MAX(statVersion)** 快照（点赞/浏览突发
+// 折叠为每计数器一条权威值——中间版本对 cn 无意义，cn 只做版本替换）→
+// HMAC 签名推送 cn 入口 POST /v1/internal/sync/stat-snapshots → 200 后把该
+// 计数器 **截至该版本** 的全部源行标 sentAt（含被折叠的低版本行——它们已被
+// 更高版本覆盖，重发永无可能改变 cn 状态）。
+//
+// 失败语义（同 sender）：任一环节抛错则本轮已推送批的确认照常落库、未推送
+// 行留待下轮（sentAt 为空即重发）；cn 侧「仅更高 statVersion 替换」保证重试
+// 幂等——「cn 已应用、ov 未收到 200」的重复回传在 cn 全部判旧（accepted=0），
+// A27 不重复累计。cn 侧 like_count/viewCountReplica 是替换式副本，本发送器
+// 从不读 cn 库（数据面单向，与 reconciler 的 cn 经 HTTP 原则一致）。
+import type { PrismaClient } from '@photo/db-ov';
+import type { components } from '@photo/contracts';
+import {
+  SYNC_SIGNATURE_HEADER,
+  SYNC_TIMESTAMP_HEADER,
+  signSyncBody,
+} from '@photo/contracts';
+import type { Logger } from 'pino';
+
+/** openapi StatSnapshot（线格式）。 */
+export type StatSnapshotWire = components['schemas']['StatSnapshot'];
+
+/** stat-snapshots 响应（openapi StatSnapshotResponse）。 */
+interface StatSnapshotResponse {
+  accepted: number;
+}
+
+/** outbox_stat_events 行（发送器只读关心的列）。 */
+type StatEventRow = {
+  id: string;
+  kind: string;
+  albumId: string;
+  photoId: string | null;
+  total: number;
+  statVersion: number;
+  createdAt: Date;
+};
+
+export interface StatReturnSenderOptions {
+  /** ov 库客户端（回传队列取数 + sentAt 回写）。 */
+  prismaOv: PrismaClient;
+  /** cn 入口基址（如 http://127.0.0.1:8080，不含路径）；端点路径本模块拼接。 */
+  cnApiUrl: string;
+  /** HMAC 密钥（与接收端 InternalSyncGuard 同一 SYNC_INTERNAL_KEY 值）。 */
+  internalKey: string;
+  /** 单批快照上限（聚合后；默认 100，与 SYNC_BATCH_SIZE 量级对齐）。 */
+  batchSize?: number;
+  /** 单次 HTTP 超时毫秒（默认 30s）。 */
+  timeoutMs?: number;
+  logger?: Logger;
+}
+
+/** 单轮 runReturnOnce 的观测口径（main 循环打点；测试用于断言折叠/幂等）。 */
+export interface StatReturnStats {
+  /** 取到的未发送源行数（聚合前）。 */
+  picked: number;
+  /** 聚合后实际发送的快照条数（=Σ 各批快照数）。 */
+  snapshots: number;
+  /** cn 实际应用（新版本生效）的快照条数；重复回传恒 0（A27 幂等锚）。 */
+  accepted: number;
+}
+
+const DEFAULT_BATCH_SIZE = 100;
+const DEFAULT_TIMEOUT_MS = 30_000;
+
+/** cn 侧 stat-snapshots 端点路径（全局前缀 v1 + InternalSyncGuard 域）。 */
+const STAT_SNAPSHOTS_PATH = '/v1/internal/sync/stat-snapshots';
+
+/** 快照来源标识（契约 StatSnapshot.source；本发送器唯一取值）。 */
+const SOURCE = 'ov';
+
+/** 聚合键：likes 按照片、views 按相册（与 outbox_stat_events 写入口径一致）。 */
+function collapseKey(row: StatEventRow): string {
+  return row.kind === 'likes'
+    ? `likes:${row.photoId ?? ''}`
+    : `views:${row.albumId}`;
+}
+
+export class StatReturnSender {
+  private readonly batchSize: number;
+  private readonly timeoutMs: number;
+  private readonly endpoint: string;
+
+  constructor(private readonly options: StatReturnSenderOptions) {
+    this.batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE;
+    this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+    this.endpoint = `${options.cnApiUrl.replace(/\/+$/, '')}${STAT_SNAPSHOTS_PATH}`;
+  }
+
+  /** 单轮回传：取未发送行 → 聚合折叠 → 分批签名推送 → 按批回写 sentAt。 */
+  async runReturnOnce(): Promise<StatReturnStats> {
+    const rows = await this.options.prismaOv.outbox_stat_events.findMany({
+      where: { sentAt: null },
+      orderBy: [{ createdAt: 'asc' }, { id: 'asc' }],
+      select: {
+        id: true,
+        kind: true,
+        albumId: true,
+        photoId: true,
+        total: true,
+        statVersion: true,
+        createdAt: true,
+      },
+    });
+    if (rows.length === 0) {
+      return { picked: 0, snapshots: 0, accepted: 0 };
+    }
+
+    // 折叠突发：同计数器只发最高版本（total 恒为该版本权威快照值）。
+    // statVersion 随每次点赞/计入单调 +1（photos_ov.countVersion /
+    // album_stats.statVersion），同键并列版本理论不出现；防御口径取「并列时
+    // 保留后见行」（createdAt 序在后）——同版本同 total，无语义差。
+    const collapsed = new Map<string, StatEventRow>();
+    for (const row of rows) {
+      const key = collapseKey(row);
+      const existing = collapsed.get(key);
+      if (
+        existing === undefined ||
+        row.statVersion >= existing.statVersion
+      ) {
+        collapsed.set(key, row);
+      }
+    }
+    const latest = [...collapsed.values()];
+
+    let accepted = 0;
+    for (let offset = 0; offset < latest.length; offset += this.batchSize) {
+      const chunk = latest.slice(offset, offset + this.batchSize);
+      const response = await this.push(chunk);
+      accepted += response.accepted;
+      await this.markSent(chunk);
+    }
+    this.options.logger?.debug(
+      { picked: rows.length, snapshots: latest.length, accepted },
+      'stat snapshots returned',
+    );
+    return { picked: rows.length, snapshots: latest.length, accepted };
+  }
+
+  /**
+   * 回写确认：对每个聚合后的快照，把该计数器**截至其版本**的未发送源行全部
+   * 标 sentAt（含被折叠的低版本行）。取数与回写之间新落库的行版本必然更高
+   * （单调自增），不会被误标；cn 200 = 整批已受理（应用或幂等判旧）。
+   */
+  private async markSent(chunk: StatEventRow[]): Promise<void> {
+    const sentAt = new Date();
+    for (const row of chunk) {
+      if (row.kind === 'likes' && row.photoId !== null) {
+        await this.options.prismaOv.outbox_stat_events.updateMany({
+          where: {
+            kind: 'likes',
+            photoId: row.photoId,
+            statVersion: { lte: row.statVersion },
+            sentAt: null,
+          },
+          data: { sentAt },
+        });
+      } else if (row.kind === 'views') {
+        await this.options.prismaOv.outbox_stat_events.updateMany({
+          where: {
+            kind: 'views',
+            albumId: row.albumId,
+            statVersion: { lte: row.statVersion },
+            sentAt: null,
+          },
+          data: { sentAt },
+        });
+      }
+    }
+  }
+
+  /**
+   * 签名推送：同一 JSON.stringify 产物既作签名输入也作请求体（rawBody 验签
+   * 字节一致，与 sender.push / reconciler.signedPost 同约定）。非 2xx / 响应
+   * 形状异常均抛错（调用方留待下轮重试）。
+   */
+  private async push(
+    snapshots: StatEventRow[],
+  ): Promise<StatSnapshotResponse> {
+    const body = {
+      snapshots: snapshots.map((row): StatSnapshotWire => {
+        if (row.kind === 'likes' && row.photoId !== null) {
+          return {
+            photoId: row.photoId,
+            albumId: row.albumId,
+            kind: 'likes',
+            total: row.total,
+            statVersion: row.statVersion,
+            source: SOURCE,
+            confirmedAt: row.createdAt.toISOString(),
+          };
+        }
+        return {
+          albumId: row.albumId,
+          kind: 'views',
+          total: row.total,
+          statVersion: row.statVersion,
+          source: SOURCE,
+          confirmedAt: row.createdAt.toISOString(),
+        };
+      }),
+    };
+    const rawBody = JSON.stringify(body);
+    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(
+        `stat-snapshots 请求失败（${this.endpoint}）：${err instanceof Error ? err.message : String(err)}`,
+      );
+    }
+    if (!response.ok) {
+      const text = await response.text().catch(() => '');
+      throw new Error(
+        `stat-snapshots 非 2xx（${this.endpoint}）：status=${response.status} body=${text.slice(0, 500)}`,
+      );
+    }
+    const data: unknown = await response.json();
+    if (
+      typeof data !== 'object' ||
+      data === null ||
+      typeof (data as StatSnapshotResponse).accepted !== 'number'
+    ) {
+      throw new Error('stat-snapshots 响应形状异常（accepted 缺失）');
+    }
+    return data as StatSnapshotResponse;
+  }
+}