photo-live/services/api/src/sync/sync.service.ts

// services/api/src/sync/sync.service.ts —— cn 侧内部同步服务（P2 Task 3：
// bootstrap/reconcile）+ Task 4 digest（每相册实体计数摘要）。共同设计决策
// （与计划文档 §2A「首次全量快照携带水位」对齐，登记为 Task 3 设计）：
//
// · bootstrap = 「把当前状态快照事件**写回 outbox**」，而不是另建拉取通道——
//   ov 冷启动时经本端点触发 cn 入队，随后由常规发送器（services/sync）批量
//   HMAC 推送、ov 按既有 content-batches 幂等/版本逻辑应用。好处：快照与增量
//   走同一条已被验证的管道（无第二条拉取路径），ov 侧无需新增消费代码。
//   重复 bootstrap 幂等：ov 对**同版本新 eventId** 事件做当前态重断言应用
//   （Task 4 起 sync-inbox 仅拒绝更旧版本；同一版本 = 同一状态，重放无副作用，
//   且正是 digest→bootstrap 补漏通道修复 ov 侧缺失副本行的机制）。
//   eventCount=入队事件数，watermark=入队事件的最高 entityVersion（契约语义
//   「后续增量对账的 sinceVersion 起点」的全局保守值；reconcile 的 sinceVersion
//   按 per-entity 版本过滤，见下）。
//
// · reconcile = 按 entityType 返回「当前状态」事件清单（entityVersion >
//   sinceVersion 的实体，升序、limit 截断）——数据源是 cn 权威表而非 outbox
//   历史流水（快照即最终状态，ov 重放即补齐；outbox 中的中间版本无需重放，
//   高版本载荷已包含最终状态）。eventId 为确定性合成 ID（recon:<type>:<id>:
//   <version>）：与真实 outbox eventId（cuid）永不冲突，同一 (实体,版本) 的
//   合成事件无论重放多少次在 ov 侧都幂等（eventId 唯一锚）。
//   已知边界（Task 4 收敛）：硬删除实体（category 墓碑后行已删）不再出现在
//   清单中——错过的墓碑经 digest 计数比对 → 该相册 bootstrap 重发 album.updated
//   （携带分类全量清单，ov 整体替换 categories）收敛。
//
// · digest（Task 4）= 每相册 {categoryCount, publishedPhotoCount, photoCount,
//   assetCount} 权威计数。ov 与副本计数比对（照片取区间语义，见 digest 注释）
//   → 不一致相册触发 bootstrap。
import { Injectable } from '@nestjs/common';
import type { components } from '@photo/contracts';
import type { Prisma } from '@photo/db-cn';
import { PrismaService } from '../prisma/prisma.service';
import type { StatSnapshotDto } from './dto';
import {
  type AlbumCategoryInput,
  buildAlbumUpdatedPayload,
  buildAssetUpdatedPayload,
  buildCategoryUpdatedPayload,
  buildPhotoPublishedPayload,
} from './payloads';

export type BootstrapResponse = components['schemas']['BootstrapResponse'];
export type ReconcileResponse = components['schemas']['ReconcileResponse'];
export type DigestResponse = components['schemas']['DigestResponse'];
export type StatSnapshotResponse = components['schemas']['StatSnapshotResponse'];
export type SyncEventWire = components['schemas']['SyncEvent'];

/** 线格式 schema 版本（契约 SyncEvent.schemaVersion 当前为 1）。 */
const SCHEMA_VERSION = 1;

/** 合成 eventId 前缀（与 cuid 分布不重叠，且可读可溯源）。 */
const RECON_EVENT_ID_PREFIX = 'recon';

@Injectable()
export class SyncService {
  constructor(private readonly prisma: PrismaService) {}

  /**
   * 全量快照引导：单事务读当前状态（album→categories→assets→photos），
   * 以同一套载荷构建器生成快照事件并写入 outbox（priority=normal，由发送器
   * 常规批量投递）。albumId 指定不存在的相册 → 空快照（eventCount=0），不报
   * 404——契约未定义 404，且对幂等重试语义更友好。
   */
  async bootstrap(albumId?: string): Promise<BootstrapResponse> {
    const occurredAt = new Date();
    const result = await this.prisma.$transaction(async (tx) => {
      const albums = await tx.albums.findMany({
        where: albumId === undefined ? {} : { id: albumId },
      });
      const albumIds = albums.map((a) => a.id);
      const albumTranslations =
        albumIds.length === 0
          ? []
          : await tx.album_translations.findMany({
              where: { albumId: { in: albumIds } },
            });
      const categories =
        albumIds.length === 0
          ? []
          : await tx.categories.findMany({
              where: { albumId: { in: albumIds } },
            });
      const categoryTranslations =
        categories.length === 0
          ? []
          : await tx.category_translations.findMany({
              where: { categoryId: { in: categories.map((c) => c.id) } },
            });
      const assets =
        albumIds.length === 0
          ? []
          : await tx.album_assets.findMany({
              where: { albumId: { in: albumIds }, state: 'effective' },
            });
      // 未发布（unpublished）照片对 ov 不可见语义，不入快照。
      const photos =
        albumIds.length === 0
          ? []
          : await tx.photos.findMany({
              where: {
                albumId: { in: albumIds },
                visibility: { in: ['published', 'hidden', 'deleted'] },
              },
            });
      const variants =
        photos.length === 0
          ? []
          : await tx.photo_variants.findMany({
              where: { photoId: { in: photos.map((p) => p.id) } },
            });

      const translationsByAlbum = new Map<string, typeof albumTranslations>();
      for (const row of albumTranslations) {
        const bucket = translationsByAlbum.get(row.albumId);
        if (bucket === undefined) {
          translationsByAlbum.set(row.albumId, [row]);
        } else {
          bucket.push(row);
        }
      }
      const translationsByCategory = new Map<
        string,
        typeof categoryTranslations
      >();
      for (const row of categoryTranslations) {
        const bucket = translationsByCategory.get(row.categoryId);
        if (bucket === undefined) {
          translationsByCategory.set(row.categoryId, [row]);
        } else {
          bucket.push(row);
        }
      }
      const variantsByPhoto = new Map<string, typeof variants>();
      for (const row of variants) {
        const bucket = variantsByPhoto.get(row.photoId);
        if (bucket === undefined) {
          variantsByPhoto.set(row.photoId, [row]);
        } else {
          bucket.push(row);
        }
      }
      // album.updated 快照载荷的分类全量清单（Task 4：ov 整体替换 categories
      // 的收敛原语）——分类行与其翻译行按 albumId 配对。
      const categoryInputsByAlbum = new Map<string, AlbumCategoryInput[]>();
      for (const category of categories) {
        const bucket = categoryInputsByAlbum.get(category.albumId);
        const input: AlbumCategoryInput = {
          category,
          translations: translationsByCategory.get(category.id) ?? [],
        };
        if (bucket === undefined) {
          categoryInputsByAlbum.set(category.albumId, [input]);
        } else {
          bucket.push(input);
        }
      }

      const rows: Prisma.outbox_eventsCreateManyInput[] = [];
      let watermark = 0;
      const push = (
        type: string,
        albumIdOf: string,
        photoId: string | null,
        entityVersion: number,
        payload: unknown,
      ): void => {
        rows.push({
          type,
          albumId: albumIdOf,
          photoId: photoId ?? undefined,
          entityVersion,
          priority: 'normal',
          payload: payload as Prisma.InputJsonValue,
          occurredAt,
        });
        watermark = Math.max(watermark, entityVersion);
      };

      for (const album of albums) {
        push(
          'album.updated',
          album.id,
          null,
          album.entityVersion,
          buildAlbumUpdatedPayload(
            album,
            translationsByAlbum.get(album.id) ?? [],
            categoryInputsByAlbum.get(album.id) ?? [],
          ),
        );
      }
      for (const category of categories) {
        push(
          'category.updated',
          category.albumId,
          null,
          category.entityVersion,
          buildCategoryUpdatedPayload(
            category,
            translationsByCategory.get(category.id) ?? [],
          ),
        );
      }
      for (const asset of assets) {
        push(
          'asset.updated',
          asset.albumId,
          null,
          asset.entityVersion,
          buildAssetUpdatedPayload(asset),
        );
      }
      for (const photo of photos) {
        if (photo.visibility === 'published') {
          push(
            'photo.published',
            photo.albumId,
            photo.id,
            photo.entityVersion,
            buildPhotoPublishedPayload(
              photo,
              variantsByPhoto.get(photo.id) ?? [],
              photo.entityVersion,
            ),
          );
        } else {
          // hidden/deleted：与 photos.service 事务内事件同构的最小事实载荷。
          push(
            photo.visibility === 'hidden' ? 'photo.hidden' : 'photo.deleted',
            photo.albumId,
            photo.id,
            photo.entityVersion,
            {
              photoId: photo.id,
              albumId: photo.albumId,
              visibility: photo.visibility,
              entityVersion: photo.entityVersion,
            },
          );
        }
      }

      if (rows.length > 0) {
        await tx.outbox_events.createMany({ data: rows });
      }
      return { eventCount: rows.length, watermark };
    });
    return result;
  }

  /**
   * 对账清单：entityType 的「当前状态」事件（entityVersion > sinceVersion，
   * 升序、limit 截断）。数据源为 cn 权威表（见文件头设计说明）。
   */
  async reconcile(query: {
    entityType: string;
    albumId?: string;
    sinceVersion: number;
    limit: number;
  }): Promise<ReconcileResponse> {
    const events: SyncEventWire[] = [];
    if (query.entityType === 'album') {
      const albums = await this.prisma.albums.findMany({
        where: {
          ...(query.albumId !== undefined ? { id: query.albumId } : {}),
          entityVersion: { gt: query.sinceVersion },
        },
        orderBy: { entityVersion: 'asc' },
        take: query.limit,
      });
      const translations =
        albums.length === 0
          ? []
          : await this.prisma.album_translations.findMany({
              where: { albumId: { in: albums.map((a) => a.id) } },
            });
      const byAlbum = new Map<string, typeof translations>();
      for (const row of translations) {
        const bucket = byAlbum.get(row.albumId);
        if (bucket === undefined) {
          byAlbum.set(row.albumId, [row]);
        } else {
          bucket.push(row);
        }
      }
      // album 快照事件携带分类全量清单（与 bootstrap 同形，见文件头说明）。
      const categoryInputs = await this.categoryInputsByAlbumIds(
        albums.map((a) => a.id),
      );
      for (const album of albums) {
        events.push(
          this.toWireEvent(
            'album',
            album.id,
            album.id,
            album.entityVersion,
            false,
            buildAlbumUpdatedPayload(
              album,
              byAlbum.get(album.id) ?? [],
              categoryInputs.get(album.id) ?? [],
            ),
          ),
        );
      }
    } else if (query.entityType === 'category') {
      const categories = await this.prisma.categories.findMany({
        where: {
          ...(query.albumId !== undefined ? { albumId: query.albumId } : {}),
          entityVersion: { gt: query.sinceVersion },
        },
        orderBy: { entityVersion: 'asc' },
        take: query.limit,
      });
      const translations =
        categories.length === 0
          ? []
          : await this.prisma.category_translations.findMany({
              where: { categoryId: { in: categories.map((c) => c.id) } },
            });
      const byCategory = new Map<string, typeof translations>();
      for (const row of translations) {
        const bucket = byCategory.get(row.categoryId);
        if (bucket === undefined) {
          byCategory.set(row.categoryId, [row]);
        } else {
          bucket.push(row);
        }
      }
      for (const category of categories) {
        events.push(
          this.toWireEvent(
            'category',
            category.id,
            category.albumId,
            category.entityVersion,
            false,
            buildCategoryUpdatedPayload(
              category,
              byCategory.get(category.id) ?? [],
            ),
          ),
        );
      }
    } else if (query.entityType === 'photo') {
      const photos = await this.prisma.photos.findMany({
        where: {
          ...(query.albumId !== undefined ? { albumId: query.albumId } : {}),
          entityVersion: { gt: query.sinceVersion },
          visibility: { in: ['published', 'hidden', 'deleted'] },
        },
        orderBy: { entityVersion: 'asc' },
        take: query.limit,
      });
      const published = photos.filter((p) => p.visibility === 'published');
      const variants =
        published.length === 0
          ? []
          : await this.prisma.photo_variants.findMany({
              where: { photoId: { in: published.map((p) => p.id) } },
            });
      const byPhoto = new Map<string, typeof variants>();
      for (const row of variants) {
        const bucket = byPhoto.get(row.photoId);
        if (bucket === undefined) {
          byPhoto.set(row.photoId, [row]);
        } else {
          bucket.push(row);
        }
      }
      for (const photo of photos) {
        if (photo.visibility === 'published') {
          events.push(
            this.toWireEvent(
              'photo',
              photo.id,
              photo.albumId,
              photo.entityVersion,
              false,
              buildPhotoPublishedPayload(
                photo,
                byPhoto.get(photo.id) ?? [],
                photo.entityVersion,
              ),
            ),
          );
        } else {
          events.push(
            this.toWireEvent(
              'photo',
              photo.id,
              photo.albumId,
              photo.entityVersion,
              photo.visibility === 'deleted',
              {
                photoId: photo.id,
                albumId: photo.albumId,
                visibility: photo.visibility,
                entityVersion: photo.entityVersion,
              },
            ),
          );
        }
      }
    } else if (query.entityType === 'asset') {
      const assets = await this.prisma.album_assets.findMany({
        where: {
          ...(query.albumId !== undefined ? { albumId: query.albumId } : {}),
          state: 'effective',
          entityVersion: { gt: query.sinceVersion },
        },
        orderBy: { entityVersion: 'asc' },
        take: query.limit,
      });
      for (const asset of assets) {
        events.push(
          this.toWireEvent(
            'asset',
            asset.id,
            asset.albumId,
            asset.entityVersion,
            false,
            buildAssetUpdatedPayload(asset),
          ),
        );
      }
    }
    // slot-state 预留：恒空清单。
    return { events };
  }

  /**
   * 对账摘要（Task 4）：cn 侧每相册的实体计数（当前权威值）。ov 侧与自己的
   * 副本计数比对，不一致 → 触发该相册 bootstrap 重建（错过的墓碑/漏掉的
   * 事件由此收敛，见 sync-inbox/Reconciler 设计说明）。photoCount 拆两档：
   * publishedPhotoCount = ov **必须**持有的行数（可见照片）；photoCount =
   * published+hidden+deleted 总数 = ov 行数上限——hidden/deleted 照片若从未
   * 在 cn 侧以 published 态同步过，ov 侧合法地无行（缺席即撤下语义），故 ov
   * 计数落在 [publishedPhotoCount, photoCount] 区间内即视为一致。assetCount
   * 供观测（ov 无 asset 副本行，不参与比对）。
   */
  async digest(albumId?: string): Promise<DigestResponse> {
    const albums = await this.prisma.albums.findMany({
      where: albumId === undefined ? {} : { id: albumId },
      select: { id: true },
      orderBy: { id: 'asc' },
    });
    if (albums.length === 0) {
      return { albums: [] };
    }
    const albumIds = albums.map((a) => a.id);
    const [categoryCounts, photoCounts, assetCounts] = await Promise.all([
      this.prisma.categories.groupBy({
        by: ['albumId'],
        where: { albumId: { in: albumIds } },
        _count: { _all: true },
      }),
      this.prisma.photos.groupBy({
        by: ['albumId', 'visibility'],
        where: {
          albumId: { in: albumIds },
          visibility: { in: ['published', 'hidden', 'deleted'] },
        },
        _count: { _all: true },
      }),
      this.prisma.album_assets.groupBy({
        by: ['albumId'],
        where: { albumId: { in: albumIds }, state: 'effective' },
        _count: { _all: true },
      }),
    ]);
    const categoriesByAlbum = new Map(
      categoryCounts.map((row) => [row.albumId, row._count._all]),
    );
    const assetsByAlbum = new Map(
      assetCounts.map((row) => [row.albumId, row._count._all]),
    );
    const photosByAlbum = new Map<
      string,
      { published: number; total: number }
    >();
    for (const row of photoCounts) {
      const bucket = photosByAlbum.get(row.albumId) ?? {
        published: 0,
        total: 0,
      };
      if (row.visibility === 'published') {
        bucket.published = row._count._all;
      }
      bucket.total += row._count._all;
      photosByAlbum.set(row.albumId, bucket);
    }
    return {
      albums: albums.map(({ id }) => {
        const photos = photosByAlbum.get(id);
        return {
          albumId: id,
          categoryCount: categoriesByAlbum.get(id) ?? 0,
          publishedPhotoCount: photos?.published ?? 0,
          photoCount: photos?.total ?? 0,
          assetCount: assetsByAlbum.get(id) ?? 0,
        };
      }),
    };
  }

  /**
   * 统计快照回传接收（P2 Task 9，ov→cn）：整批单事务，逐快照做**版本替换**——
   * 仅当 cn 侧 statsVersion < snapshot.statVersion 时以权威 total 整体替换副本
   * （绝不相加），重复回传（同版本）与乱序（更低版本）均幂等忽略。这是 A27
   * 「不重复累计」的接收端锚点：total 恒为 ov 权威快照值（非增量），即使未来
   * 出现下调修正（total 变小），更高版本也照常替换——版本是唯一权威。
   *
   * 副本位：likes → photos.likeCount（比较位 photos.statsVersion）；
   * views → albums.viewCountReplica（比较位 albums.statsVersion，
   * 同时推进 statsSyncedAt=本批应用时刻，A27 展示同步时间；likes 快照的
   * photoId 在 cn 必然存在（照片源于 cn），不存在/更旧版本 → 该条不计入
   * accepted，不报错——与「旧版本忽略不计」同一幂等口径）。
   * accepted 只数**实际应用**（新版本生效）的条数（契约：旧版本被忽略的不计）。
   */
  async receiveStatSnapshots(
    snapshots: StatSnapshotDto[],
  ): Promise<StatSnapshotResponse> {
    const syncedAt = new Date();
    return this.prisma.$transaction(async (tx) => {
      let accepted = 0;
      for (const snapshot of snapshots) {
        if (snapshot.kind === 'likes' && snapshot.photoId !== undefined) {
          const applied = await tx.photos.updateMany({
            where: {
              id: snapshot.photoId,
              statsVersion: { lt: snapshot.statVersion },
            },
            data: {
              likeCount: snapshot.total,
              statsVersion: snapshot.statVersion,
            },
          });
          accepted += applied.count;
        } else if (snapshot.kind === 'views') {
          const applied = await tx.albums.updateMany({
            where: {
              id: snapshot.albumId,
              statsVersion: { lt: snapshot.statVersion },
            },
            data: {
              viewCountReplica: snapshot.total,
              statsVersion: snapshot.statVersion,
              statsSyncedAt: syncedAt,
            },
          });
          accepted += applied.count;
        }
      }
      return { accepted };
    });
  }

  /** 相册 id 集合 → 分类快照输入（分类行 + 翻译行配对；升序确定性由构建器排）。 */
  private async categoryInputsByAlbumIds(
    albumIds: string[],
  ): Promise<Map<string, AlbumCategoryInput[]>> {
    const result = new Map<string, AlbumCategoryInput[]>();
    if (albumIds.length === 0) {
      return result;
    }
    const categories = await this.prisma.categories.findMany({
      where: { albumId: { in: albumIds } },
    });
    const translations =
      categories.length === 0
        ? []
        : await this.prisma.category_translations.findMany({
            where: { categoryId: { in: categories.map((c) => c.id) } },
          });
    const translationsByCategory = new Map<
      string,
      typeof translations
    >();
    for (const row of translations) {
      const bucket = translationsByCategory.get(row.categoryId);
      if (bucket === undefined) {
        translationsByCategory.set(row.categoryId, [row]);
      } else {
        bucket.push(row);
      }
    }
    for (const category of categories) {
      const bucket = result.get(category.albumId);
      const input: AlbumCategoryInput = {
        category,
        translations: translationsByCategory.get(category.id) ?? [],
      };
      if (bucket === undefined) {
        result.set(category.albumId, [input]);
      } else {
        bucket.push(input);
      }
    }
    return result;
  }

  /** 当前状态 → 线格式事件（确定性合成 eventId，见文件头设计说明）。 */
  private toWireEvent(
    entityType: SyncEventWire['entityType'],
    entityId: string,
    albumId: string,
    entityVersion: number,
    tombstone: boolean,
    payload: unknown,
  ): SyncEventWire {
    return {
      eventId: `${RECON_EVENT_ID_PREFIX}:${entityType}:${entityId}:${entityVersion}`,
      schemaVersion: SCHEMA_VERSION,
      entityType,
      entityId,
      albumId,
      entityVersion,
      // 显式携带（生成类型必填，见 sync-inbox dto 同注）。
      tombstone,
      priority: 'normal',
      payload: payload as SyncEventWire['payload'],
    };
  }
}