photo-live/services/sync/src/stat-return.ts

// 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;
  }
}