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

// services/api/src/sync-inbox/sync-inbox.service.ts —— ov 内容事件收件箱
// （P2 Task 3：接收 + 持久化 + 应用；Task 4 起同版本重断言语义）。逐事件单事务：
//
//   ① eventId 幂等：sync_inbox 主键冲突（含并发窗口）→ duplicate，受理不计应用；
//   ② entityVersion 单调：sync_state 已应用版本 **严格大于** 事件版本 → stale，
//      标记 appliedAt（已裁决，非积压）但不投影——旧事件不能复活新状态（A26）；
//      **等于**已应用版本 → 重断言应用（Task 4 语义）：cn 侧 entityVersion 与
//      状态一一对应（每次变更事务内自增），同版本 = 同状态，重放为幂等重投影
//      ——bootstrap/reconcile 快照重发（新 eventId、同版本）正是靠它修复 ov 侧
//      缺失/漂移的副本行（digest→bootstrap 补漏通道）；正常增量管线不受影响
//      （同事件重投递走 ① 的 eventId 幂等锚，根本到不了版本比较）；
//   ③ 依赖缺失暂存：非 album 事件在 albums_ov 无所属相册行时保持 appliedAt
//      null（暂存不 applied，不投影）；album 事件应用后按 albumId 回扫暂存
//      队列补应用（相册到达即解锁该相册全部暂存事件）；
//   ④ 应用投影（版本不低于已应用）：album → albums_ov upsert（不触碰
//      bannerAssetKey/emergencyDisabled——分别归 asset.updated / 紧急停用通道
//      Task 5；categories 仅当载荷携带全量清单（bootstrap/reconcile 快照事件）
//      时整体替换——错过的 category 墓碑由此收敛，增量事件不携带则不动）；
//      category → albums_ov.categories JSON 元素 upsert/移除；photo.published →
//      photos_ov upsert（含 categoryId——Task 6 公开列表分类过滤/单图同形所需；
//      保留 likeTotal/countVersion 权威计数）；photo.hidden/
//      deleted → photos_ov.visibility 置位（行不存在 = 从未复制即不可见，缺席
//      即撤下语义）；asset → banner 投影 albums_ov.bannerAssetKey（cover/logo
//      仅推进版本，ov schema 无位）；
//   ⑤ sync_state upsert appliedVersion（watermark=false 实体行；水位行预留）。
//   ⑥ 实时转发（Task 7）：photo 事件**应用事务提交后**发布 OV_PHOTO_APPLIED
//      进程内事件 → realtime 网关向相册房间广播契约 photo.* 消息（stale/
//      duplicate/staged 不转发；从未复制即撤下的行也不转发——观众端无此照片
//      可撤，广播无意义）。
//
// 墓碑优先由 ② 保证：tombstone 事件版本必高于其前的任意旧版本，旧
// photo.published/category.updated 重放被版本比较直接拒绝。
import { Injectable } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import type { components } from '@photo/contracts';
import type {
  AlbumUpdatedPayload,
  AssetUpdatedPayload,
  CategoryDeletedPayload,
  CategoryUpdatedPayload,
  PhotoPublishedPayload,
  PhotoVisibilityChangedPayload,
} from '@photo/contracts';
import type { Prisma } from '@photo/db-ov';
// 值导入（非 import type）：构造器注入的 design:paramtypes 依赖类引用，
// type-only 导入会让 Nest 侧元数据退化为 Function 而 DI 失败（repo 约定）。
import { OvPrismaService } from '../prisma-ov/ov-prisma.module';
import { OV_PHOTO_APPLIED, type PhotoAppliedEvent } from '../realtime/ov-events';

export type ContentBatchResponse =
  components['schemas']['ContentBatchResponse'];
type SyncEventWire = components['schemas']['SyncEvent'];

/** 单事件裁决结果（响应只聚合 applied，其余内部口径）。 */
type Outcome = 'duplicate' | 'stale' | 'staged' | 'applied';

/** albums_ov.categories JSON 元素形状（ov schema 注释约定的投影结构）。 */
interface CategoryReplicaEntry {
  id: string;
  sortOrder: number;
  translations: Array<{ locale: string; name: string }>;
}

@Injectable()
export class SyncInboxService {
  constructor(
    private readonly prisma: OvPrismaService,
    private readonly events: EventEmitter2,
  ) {}

  /**
   * 批量接收：批内按序逐事件事务处理；任一事件失败抛错（整批 500）——发送器
   * 整批重试，eventId 幂等保证重试安全。acceptedEventIds 含幂等重复（契约：
   * 受理 ≠ 应用）；applied 仅统计新版本生效。
   */
  async receiveBatch(batch: SyncEventWire[]): Promise<ContentBatchResponse> {
    const acceptedEventIds: string[] = [];
    let applied = 0;
    for (const event of batch) {
      const outcome = await this.applyOne(event);
      acceptedEventIds.push(event.eventId);
      if (outcome === 'applied') {
        applied += 1;
        // 依赖解锁：相册副本落位后，回扫该相册的暂存事件（photo/category/
        // asset 都只依赖相册行；album 事件自身永不暂存，回扫无递归）。
        if (event.entityType === 'album') {
          applied += await this.sweepStaged(event.albumId);
        }
      }
    }
    return { acceptedEventIds, applied };
  }

  /** 单事件事务（见文件头 ①-⑥）。P2002 = 并发同 eventId → duplicate。 */
  async applyOne(event: SyncEventWire): Promise<Outcome> {
    let realtime: PhotoAppliedEvent | undefined;
    try {
      const outcome = await this.prisma.$transaction(async (tx) => {
        // ① eventId 幂等锚：先查后插 + 唯一约束兜底（并发窗口）。
        const seen = await tx.sync_inbox.findUnique({
          where: { eventId: event.eventId },
          select: { eventId: true },
        });
        if (seen !== null) {
          return 'duplicate' as const;
        }
        await tx.sync_inbox.create({
          data: {
            eventId: event.eventId,
            entityType: event.entityType,
            entityId: event.entityId,
            albumId: event.albumId,
            entityVersion: event.entityVersion,
            tombstone: event.tombstone === true,
            priority: event.priority,
            payload: event.payload as unknown as Prisma.InputJsonValue,
          },
        });
        const applied = await this.decideAndApply(tx, event);
        realtime = applied.realtime;
        return applied.outcome;
      });
      // ⑥ 提交后转发（$transaction resolve 即已 COMMIT；回滚路径上面已抛）。
      if (realtime !== undefined) {
        this.events.emit(OV_PHOTO_APPLIED, realtime);
      }
      return outcome;
    } catch (err) {
      // P2002 鸭子类型判别（同 albums/support.ts isUniqueViolationOn 的约定：
      // Prisma 命名空间经 @photo/db-ov 仅 type 出口，不做值导入）——本端点
      // 唯一约束只有 sync_inbox 主键，命中即并发同 eventId。
      if (
        typeof err === 'object' &&
        err !== null &&
        (err as { code?: unknown }).code === 'P2002'
      ) {
        return 'duplicate';
      }
      throw err;
    }
  }

  /**
   * ②-⑤ 裁决与应用（inbox 行已存在的共用核心）：新事件（applyOne 落库后）
   * 与暂存回扫（sweepStaged 对既有行重跑）同一出口。回扫**不能**走 applyOne
   * 整程——其 eventId 幂等检查会把暂存行判成 duplicate 而永不应用。返回
   * ⑥ 的实时转发事件（photo 事件才有；stale/staged 无）。
   */
  private async decideAndApply(
    tx: Prisma.TransactionClient,
    event: SyncEventWire,
  ): Promise<{ outcome: Outcome; realtime?: PhotoAppliedEvent }> {
    // ② 版本单调：**更旧**版本裁决即终（appliedAt 置位，不入积压）；**等于**
    // 已应用版本 → 重断言应用（同版本=同状态，幂等重投影；digest→bootstrap
    // 补漏修复缺失副本行的机制，见文件头②）。
    const state = await tx.sync_state.findUnique({
      where: {
        entityType_entityId: {
          entityType: event.entityType,
          entityId: event.entityId,
        },
      },
    });
    if (state !== null && state.appliedVersion > event.entityVersion) {
      await tx.sync_inbox.update({
        where: { eventId: event.eventId },
        data: { appliedAt: new Date() },
      });
      return { outcome: 'stale' };
    }
    // ③ 依赖检查：非 album 事件须所属相册副本已存在，否则暂存。
    if (event.entityType !== 'album') {
      const album = await tx.albums_ov.findUnique({
        where: { albumId: event.albumId },
        select: { albumId: true },
      });
      if (album === null) {
        return { outcome: 'staged' };
      }
    }
    // ④ 投影（photo 事件带回实时转发载荷）。
    const realtime = await this.project(tx, event);
    // ⑤ 版本账本。
    await tx.sync_state.upsert({
      where: {
        entityType_entityId: {
          entityType: event.entityType,
          entityId: event.entityId,
        },
      },
      update: {
        appliedVersion: event.entityVersion,
        appliedAt: new Date(),
      },
      create: {
        entityType: event.entityType,
        entityId: event.entityId,
        appliedVersion: event.entityVersion,
        watermark: false,
      },
    });
    await tx.sync_inbox.update({
      where: { eventId: event.eventId },
      data: { appliedAt: new Date() },
    });
    return { outcome: 'applied', realtime };
  }

  /** ④ 按实体类型投影到副本表（版本已由 ② 保证更高）。 */
  private async project(
    tx: Prisma.TransactionClient,
    event: SyncEventWire,
  ): Promise<PhotoAppliedEvent | undefined> {
    if (event.entityType === 'album') {
      const p = event.payload as unknown as AlbumUpdatedPayload;
      // categories 全量清单（可选，仅 bootstrap/reconcile 快照事件携带）：整体
      // 替换 albums_ov.categories——错过的 category 墓碑（cn 行已删、无事件可
      // 重放）只有全量替换才能收敛（Task 4 digest→bootstrap 补漏）。增量事件
      // 不携带 → 不触碰（分类元素归 category.* 事件管）。bannerAssetKey/
      // emergencyDisabled 永不触碰（分别归 asset.updated / 紧急停用通道）。
      const categorySnapshot = Array.isArray(p.categories)
        ? (p.categories as unknown as Prisma.InputJsonValue)
        : undefined;
      await tx.albums_ov.upsert({
        where: { albumId: event.entityId },
        create: {
          albumId: event.entityId,
          publicToken: p.publicToken,
          status: p.status,
          publicEnabled: p.publicEnabled,
          downloadEnabled: p.downloadEnabled,
          originalDownloadEnabled: p.originalDownloadEnabled,
          translations: p.translations as unknown as Prisma.InputJsonValue,
          categories: categorySnapshot ?? [],
          defaultLocale: p.defaultLocale,
          timezone: p.timezone,
          startAt: new Date(p.startAt),
          endAt: p.endAt === null ? null : new Date(p.endAt),
          entityVersion: event.entityVersion,
        },
        update: {
          publicToken: p.publicToken,
          status: p.status,
          publicEnabled: p.publicEnabled,
          downloadEnabled: p.downloadEnabled,
          originalDownloadEnabled: p.originalDownloadEnabled,
          translations: p.translations as unknown as Prisma.InputJsonValue,
          ...(categorySnapshot !== undefined
            ? { categories: categorySnapshot }
            : {}),
          defaultLocale: p.defaultLocale,
          timezone: p.timezone,
          startAt: new Date(p.startAt),
          endAt: p.endAt === null ? null : new Date(p.endAt),
          entityVersion: event.entityVersion,
        },
      });
      return undefined;
    }
    if (event.entityType === 'category') {
      const album = await tx.albums_ov.findUniqueOrThrow({
        where: { albumId: event.albumId },
      });
      const entries = (
        Array.isArray(album.categories) ? album.categories : []
      ) as unknown as CategoryReplicaEntry[];
      if (event.tombstone === true) {
        const p = event.payload as unknown as CategoryDeletedPayload;
        await tx.albums_ov.update({
          where: { albumId: event.albumId },
          data: {
            // 接口类型无 Json 输入所需索引签名（同 outbox 写入侧约定），结构化
            // 拷贝落 Json 列。
            categories: entries.filter(
              (c) => c.id !== p.categoryId,
            ) as unknown as Prisma.InputJsonValue,
          },
        });
        return undefined;
      }
      const p = event.payload as unknown as CategoryUpdatedPayload;
      const entry: CategoryReplicaEntry = {
        id: p.categoryId,
        sortOrder: p.sortOrder,
        translations: p.translations,
      };
      const index = entries.findIndex((c) => c.id === p.categoryId);
      if (index >= 0) {
        entries[index] = entry;
      } else {
        entries.push(entry);
      }
      await tx.albums_ov.update({
        where: { albumId: event.albumId },
        data: { categories: entries as unknown as Prisma.InputJsonValue },
      });
      return undefined;
    }
    if (event.entityType === 'photo') {
      const payload = event.payload as unknown as Partial<
        PhotoPublishedPayload & PhotoVisibilityChangedPayload
      >;
      // visibility 推导：hidden/deleted 载荷显式携带；published 载荷缺省即发布。
      const visibility =
        payload.visibility ?? (event.tombstone === true ? 'deleted' : 'published');
      const existing = await tx.photos_ov.findUnique({
        where: { photoId: event.entityId },
        select: { photoId: true },
      });
      if (existing === null) {
        if (visibility !== 'published') {
          // 墓碑/隐藏先于任何 published 到达且副本行不存在：缺席即撤下语义，
          // 无行可改——版本账本照常推进（旧 published 之后到会被版本比较拒绝）。
          return undefined;
        }
        const p = payload as PhotoPublishedPayload;
        await tx.photos_ov.create({
          data: {
            photoId: event.entityId,
            albumId: event.albumId,
            categoryId: p.categoryId ?? null,
            originalObjectKey: p.objectKey,
            thumbObjectKey: p.thumbKey,
            browseObjectKey: p.browseKey,
            width: p.width,
            height: p.height,
            visibility,
            publishedAt: new Date(p.publishedAt),
            entityVersion: event.entityVersion,
          },
        });
        // ⑥ 实时转发载荷（photo.published）。
        return {
          kind: 'published',
          photoId: event.entityId,
          albumId: event.albumId,
          entityVersion: event.entityVersion,
          publishedAt: p.publishedAt,
        };
      }
      // update 分支保留 likeTotal/countVersion（ov 权威计数，Task 7/9）。
      const data: Prisma.photos_ovUpdateInput = {
        visibility,
        entityVersion: event.entityVersion,
      };
      if (visibility === 'published') {
        const p = payload as PhotoPublishedPayload;
        data.categoryId = p.categoryId ?? null;
        data.originalObjectKey = p.objectKey;
        data.thumbObjectKey = p.thumbKey;
        data.browseObjectKey = p.browseKey;
        data.width = p.width;
        data.height = p.height;
        data.publishedAt = new Date(p.publishedAt);
      }
      await tx.photos_ov.update({
        where: { photoId: event.entityId },
        data,
      });
      // ⑥ 实时转发载荷：published 直映；hidden/deleted 按契约消息；其余
      // visibility 字符串（unpublished 等防御路径）按 hidden 语义转发——
      // 观众端的正确反应都是「即刻移出列表」。
      if (visibility === 'published') {
        const p = payload as PhotoPublishedPayload;
        return {
          kind: 'published',
          photoId: event.entityId,
          albumId: event.albumId,
          entityVersion: event.entityVersion,
          publishedAt: p.publishedAt,
        };
      }
      return {
        kind: visibility === 'deleted' ? 'deleted' : 'hidden',
        photoId: event.entityId,
        albumId: event.albumId,
        entityVersion: event.entityVersion,
      };
    }
    if (event.entityType === 'asset') {
      const p = event.payload as unknown as AssetUpdatedPayload;
      if (p.kind === 'banner') {
        await tx.albums_ov.update({
          where: { albumId: event.albumId },
          data: { bannerAssetKey: p.objectKey },
        });
      }
      // cover/logo：ov schema 无投影位（YAGNI，schema 注释），仅推进版本账本。
      return undefined;
    }
    // slot-state 预留：受理即 applied（无投影），版本账本照常推进。
    return undefined;
  }

  /** album 应用后回扫该相册暂存事件（receivedAt 升序），返回补应用数。 */
  private async sweepStaged(albumId: string): Promise<number> {
    const staged = await this.prisma.sync_inbox.findMany({
      where: { appliedAt: null, albumId },
      orderBy: [{ receivedAt: 'asc' }, { eventId: 'asc' }],
    });
    let applied = 0;
    for (const row of staged) {
      const event: SyncEventWire = {
        eventId: row.eventId,
        schemaVersion: 1,
        entityType: row.entityType as SyncEventWire['entityType'],
        entityId: row.entityId,
        albumId: row.albumId ?? albumId,
        entityVersion: row.entityVersion,
        tombstone: row.tombstone,
        priority: row.priority as SyncEventWire['priority'],
        payload: row.payload as unknown as SyncEventWire['payload'],
      };
      // 行已存在（eventId 即本行主键），走 ②-⑥ 核心而非 applyOne 整程
      // （幂等检查会把回扫误判 duplicate，见 decideAndApply 注释）。
      const result = await this.prisma.$transaction((tx) =>
        this.decideAndApply(tx, event),
      );
      if (result.outcome === 'applied') {
        applied += 1;
        // ⑥ 暂存补应用同样提交后转发（依赖相册到达的迟到照片事件）。
        if (result.realtime !== undefined) {
          this.events.emit(OV_PHOTO_APPLIED, result.realtime);
        }
      }
    }
    return applied;
  }
}