photo-live/services/sync/src/reconciler.ts

// services/sync/src/reconciler.ts —— ov 补漏对账调度器（P2 Task 4）。
// runOnce() 两阶段，每 SYNC_RECONCILE_INTERVAL_MS 一轮（main 循环驱动）：
//
//   阶段一「当前态清扫」：对 album/category/photo/asset 四类逐类分页拉取 cn
//   reconcile 清单（**当前状态事件**，非版本清单——Task 4 设计决策：重放走
//   既有 content-batches 管道即幂等，无需第二条拉取通道），直接签名推送到 ov
//   content-batches。sinceVersion 取 0（全量当前态）而非「per-type 已应用最大
//   版本」——cn reconcile 以**类型级全局版本阈值**过滤，类型内单实体滞后
//   （如 A 实体已 v9、B 实体 v3→v4）会被 max 掩蔽；0 则由 ov 的 eventId 幂等
//   + 版本单调把「已追平的实体」全部判重/判旧（确定性 recon:<type>:<id>:<ver>
//   合成 eventId，重扫即纯重复检查），任何滞后/新实体必然被补上。仿真规模
//   （单册数百照片）下每轮开销可忽略；真实地域部署若需缩量，改点在 cn 侧
//   提供 per-entity 水位比对（登记为 P5 演进项，不在 P2）。
//
//   阶段二「digest 比对 → bootstrap 重建」：拉取 cn digest（每相册权威计数：
//   categoryCount / publishedPhotoCount..photoCount 区间 / assetCount）与 ov
//   副本计数比对。不一致的相册 → 触发 cn bootstrap {albumId}（当前状态快照
//   写回 outbox，由常驻发送器投递）。收敛机制依赖两条 Task 4 语义：
//     · sync-inbox 对**同版本新 eventId** 事件做当前态重断言应用（cn 的
//       entityVersion 与状态一一对应）→ 修复 ov 缺失的 photos_ov 行；
//     · bootstrap/reconcile 的 album.updated 载荷携带分类全量清单，ov 整体
//       替换 albums_ov.categories → 修复**错过的 category 墓碑**（cn 行已删，
//       任何「重放现存实体」通道都无法表达的删除语义，Task 3 已登记边界）。
//   照片计数取区间口径：hidden/deleted 照片若从未以 published 态同步过，ov
//   合法地无行（缺席即撤下），故 ov 行数 ∈ [publishedPhotoCount, photoCount]
//   即一致。ov 有而 cn 无的相册（cn 行已删，不可经 bootstrap 重建）仅告警
//   （P4a 管理页消费 sync_state/sync_inbox 的延迟与积压口径）。
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';

type SyncEventWire = components['schemas']['SyncEvent'];
type DigestResponse = components['schemas']['DigestResponse'];

/** 线格式端点路径（全局前缀 v1 + InternalSyncGuard 域；与 sender 同约定）。 */
const RECONCILE_PATH = '/v1/internal/sync/reconcile';
const DIGEST_PATH = '/v1/internal/sync/digest';
const BOOTSTRAP_PATH = '/v1/internal/sync/bootstrap';
const CONTENT_BATCHES_PATH = '/v1/internal/sync/content-batches';

/** 清扫覆盖的实体类型（slot-state 预留，无当前态可拉）。 */
const SWEEP_ENTITY_TYPES = ['album', 'category', 'photo', 'asset'] as const;

const DEFAULT_RECONCILE_LIMIT = 500; // 任务口径单页上限（契约上限 1000）
const DEFAULT_PUSH_BATCH_SIZE = 100; // content-batches 契约单批上限
const DEFAULT_TIMEOUT_MS = 30_000;
/** 分页保险丝：单类型最多拉 100 页（5 万实体）——防响应形状异常导致的死循环。 */
const MAX_PAGES_PER_TYPE = 100;

export interface SyncReconcilerOptions {
  /** ov 库客户端（副本计数比对直读；cn 侧数据一律经内部 HTTP，不直连 cn 库）。 */
  prismaOv: PrismaClient;
  /** cn 入口基址（如 http://127.0.0.1:8080，不含路径）。 */
  cnApiUrl: string;
  /** ov 入口基址（重放事件推送目标）。 */
  ovApiUrl: string;
  /** HMAC 密钥（与两端 InternalSyncGuard 同一 SYNC_INTERNAL_KEY 值）。 */
  internalKey: string;
  /** reconcile 单页上限（默认 500）。 */
  reconcileLimit?: number;
  /** content-batches 单批上限（默认契约 100）。 */
  pushBatchSize?: number;
  /** 单次 HTTP 超时毫秒（默认 30s）。 */
  timeoutMs?: number;
  logger?: Logger;
}

/** 单轮 runOnce 的观测口径（main 循环打点；测试用于断言追赶/补漏/稳态）。 */
export interface ReconcileStats {
  /** 阶段一从 cn 拉到的当前态事件数（四类型合计）。 */
  sweptEvents: number;
  /** 阶段一重放中 ov 实际应用（新版本/同版本重断言）的条数。 */
  replayApplied: number;
  /** 阶段二 cn digest 相册数。 */
  digestAlbums: number;
  /** 计数不一致 → 已触发 bootstrap 的相册（ healed by sender 投递）。 */
  mismatchedAlbums: string[];
  /** ov 存在而 cn 无的相册（cn 行已删——不可 bootstrap 重建，仅告警登记）。 */
  ovOrphanAlbums: string[];
}

/** ov 侧每相册副本计数（albums_ov.categories 为 JSON，行数小，直接取回计算）。 */
interface OvAlbumCounts {
  categoryCount: number;
  photoCount: number;
}

export class SyncReconciler {
  private readonly reconcileLimit: number;
  private readonly pushBatchSize: number;
  private readonly timeoutMs: number;
  private readonly cnBase: string;
  private readonly ovBase: string;

  constructor(private readonly options: SyncReconcilerOptions) {
    this.reconcileLimit = options.reconcileLimit ?? DEFAULT_RECONCILE_LIMIT;
    this.pushBatchSize = options.pushBatchSize ?? DEFAULT_PUSH_BATCH_SIZE;
    this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
    this.cnBase = options.cnApiUrl.replace(/\/+$/, '');
    this.ovBase = options.ovApiUrl.replace(/\/+$/, '');
  }

  /** 单轮对账：阶段一清扫重放 + 阶段二 digest 比对触发 bootstrap。 */
  async runOnce(): Promise<ReconcileStats> {
    const { sweptEvents, replayApplied } = await this.sweep();
    const digest = await this.digestCompare();
    const stats: ReconcileStats = {
      sweptEvents,
      replayApplied,
      digestAlbums: digest.digestAlbums,
      mismatchedAlbums: digest.mismatchedAlbums,
      ovOrphanAlbums: digest.ovOrphanAlbums,
    };
    this.options.logger?.info(
      {
        sweptEvents,
        replayApplied,
        digestAlbums: stats.digestAlbums,
        mismatched: stats.mismatchedAlbums.length,
        ovOrphans: stats.ovOrphanAlbums.length,
      },
      'reconcile cycle done',
    );
    return stats;
  }

  /**
   * 阶段一：逐类型分页拉取 cn 当前态事件并签名推送到 ov content-batches。
   * 任一类型失败抛错（整轮中止，下轮重来——幂等安全）。
   */
  private async sweep(): Promise<{
    sweptEvents: number;
    replayApplied: number;
  }> {
    let sweptEvents = 0;
    let replayApplied = 0;
    for (const entityType of SWEEP_ENTITY_TYPES) {
      for (let page = 0; page < MAX_PAGES_PER_TYPE; page += 1) {
        const response = await this.signedPost<{
          events: SyncEventWire[];
        }>(this.cnBase, RECONCILE_PATH, {
          entityType,
          sinceVersion: 0,
          limit: this.reconcileLimit,
        });
        const events = response.events;
        sweptEvents += events.length;
        for (
          let offset = 0;
          offset < events.length;
          offset += this.pushBatchSize
        ) {
          const chunk = events.slice(offset, offset + this.pushBatchSize);
          const batch = await this.signedPost<{
            acceptedEventIds: string[];
            applied: number;
          }>(this.ovBase, CONTENT_BATCHES_PATH, { batch: chunk });
          replayApplied += batch.applied;
        }
        if (events.length < this.reconcileLimit) {
          break;
        }
      }
    }
    return { sweptEvents, replayApplied };
  }

  /**
   * 阶段二：cn digest 与 ov 副本计数比对；不一致相册触发 cn bootstrap
   * （快照写回 outbox，由常驻发送器投递——本调度器不直接投递 bootstrap 事件，
   * 与发送器职责分离）。返回比对口径供打点/测试断言。
   */
  private async digestCompare(): Promise<{
    digestAlbums: number;
    mismatchedAlbums: string[];
    ovOrphanAlbums: string[];
  }> {
    const digest = await this.signedPost<DigestResponse>(
      this.cnBase,
      DIGEST_PATH,
      {},
    );
    const cnAlbums = new Map(digest.albums.map((row) => [row.albumId, row]));
    const ovCounts = await this.ovAlbumCounts();
    const mismatchedAlbums: string[] = [];
    for (const cnRow of digest.albums) {
      const ov = ovCounts.get(cnRow.albumId);
      if (
        ov === undefined ||
        ov.categoryCount !== cnRow.categoryCount ||
        ov.photoCount < cnRow.publishedPhotoCount ||
        ov.photoCount > cnRow.photoCount
      ) {
        mismatchedAlbums.push(cnRow.albumId);
      }
    }
    const ovOrphanAlbums = [...ovCounts.keys()]
      .filter((albumId) => !cnAlbums.has(albumId))
      .sort();
    if (ovOrphanAlbums.length > 0) {
      // cn 行已删（或 ov 指向了错误源）：bootstrap 无法表达「删除相册」，
      // 不自动处置——告警登记，人工/P4a 管理通道跟进。
      this.options.logger?.error(
        { ovOrphanAlbums },
        'digest: ov 存在 cn 已不存在的相册副本（不可自动收敛，登记跟进）',
      );
    }
    for (const albumId of mismatchedAlbums) {
      await this.signedPost(this.cnBase, BOOTSTRAP_PATH, { albumId });
    }
    if (mismatchedAlbums.length > 0) {
      this.options.logger?.warn(
        { mismatchedAlbums },
        'digest: 副本计数不一致，已触发受影响相册 bootstrap 重建',
      );
    }
    return {
      digestAlbums: digest.albums.length,
      mismatchedAlbums,
      ovOrphanAlbums,
    };
  }

  /** ov 每相册副本计数：albums_ov 行（categories JSON 长度）+ photos_ov 分组数。 */
  private async ovAlbumCounts(): Promise<Map<string, OvAlbumCounts>> {
    const [albumRows, photoRows] = await Promise.all([
      this.options.prismaOv.albums_ov.findMany({
        select: { albumId: true, categories: true },
        orderBy: { albumId: 'asc' },
      }),
      this.options.prismaOv.photos_ov.groupBy({
        by: ['albumId'],
        _count: { _all: true },
      }),
    ]);
    const photosByAlbum = new Map(
      photoRows.map((row) => [row.albumId, row._count._all]),
    );
    const counts = new Map<string, OvAlbumCounts>();
    for (const row of albumRows) {
      counts.set(row.albumId, {
        categoryCount: Array.isArray(row.categories)
          ? row.categories.length
          : 0,
        photoCount: photosByAlbum.get(row.albumId) ?? 0,
      });
    }
    return counts;
  }

  /**
   * 签名 POST：同一 JSON.stringify 产物既作签名输入也作请求体（rawBody 验签
   * 字节一致，与 sender.push 同约定）；非 2xx 抛错（调用方整轮重试语义）。
   */
  private async signedPost<T>(
    base: string,
    path: string,
    body: unknown,
  ): Promise<T> {
    const rawBody = JSON.stringify(body);
    const timestamp = Math.floor(Date.now() / 1000).toString();
    let response: Response;
    const endpoint = `${base}${path}`;
    try {
      response = await fetch(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(
        `${path} 请求失败（${endpoint}）：${err instanceof Error ? err.message : String(err)}`,
      );
    }
    if (!response.ok) {
      const text = await response.text().catch(() => '');
      throw new Error(
        `${path} 非 2xx（${endpoint}）：status=${response.status} body=${text.slice(0, 500)}`,
      );
    }
    return (await response.json()) as T;
  }
}