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

// services/sync/src/sender.ts —— cn→ov outbox 发送器（P2 Task 3）。
// 单次推送 runOnce()：从 cn 库 outbox_events 取 deliveredAt 为空的事件一批，
// 转内部同步线格式（openapi SyncEvent）后 HMAC 签名推送 ov 入口
// POST /v1/internal/sync/content-batches；响应 acceptedEventIds 即「受理」锚点
// （含幂等重复），据此回写 deliveredAt。
//
// 高优先级排序：**显式两段取数**（先 high 后其余，段内 occurredAt 升序）而非
// ORDER BY priority DESC——'high' < 'normal'（字典序 h<n），DESC 反而把 normal
// 排前；两段取数不依赖任何字符串序假设，新增优先级值也不失效。间隔轮询模型下
// high 事件最坏等待一个 SYNC_PUSH_INTERVAL_MS（默认 2s，D11「high 不受间隔
// 约束」的落地取舍：2s 上限可接受，专路即时唤醒属 YAGNI）。
//
// 失败语义：任一环节抛错则本 tick 整批不标记、下轮重推（eventId 幂等保证重试
// 安全）；退避由 main 循环负责（指数递增封顶），本模块无内部重试。
import type { PrismaClient } from '@photo/db-cn';
import type { components } from '@photo/contracts';
import {
  type OutboxEvent,
  SYNC_SIGNATURE_HEADER,
  SYNC_TIMESTAMP_HEADER,
  isTombstoneOutboxType,
  signSyncBody,
  syncEntityTypeOfOutboxType,
} from '@photo/contracts';
import type { Logger } from 'pino';

/** openapi SyncEvent（线格式；payload 为实体类型相关载荷对象）。 */
export type SyncEventWire = components['schemas']['SyncEvent'];

/** ov content-batches 响应（openapi ContentBatchResponse）。 */
interface ContentBatchResponse {
  acceptedEventIds: string[];
  applied: number;
}

/** outbox 行（发送器只读关心的列）。 */
type OutboxRow = {
  id: string;
  type: string;
  albumId: string;
  photoId: string | null;
  entityVersion: number;
  priority: string;
  payload: unknown;
  occurredAt: Date;
};

export interface OutboxSenderOptions {
  prisma: PrismaClient;
  /** ov 入口基址（如 http://127.0.0.1:8081，不含路径）；端点路径本模块拼接。 */
  ovApiUrl: string;
  /** HMAC 密钥（与接收端 InternalSyncGuard 同一 SYNC_INTERNAL_KEY 值）。 */
  internalKey: string;
  /** 单批上限（默认契约 SYNC_BATCH_SIZE）。 */
  batchSize?: number;
  /** 单次 HTTP 超时毫秒（默认 30s）。 */
  timeoutMs?: number;
  logger?: Logger;
}

/** 单次 runOnce 的观测口径（main 循环打点；测试用于排空/排序判断）。 */
export interface PushStats {
  /** 本批取到的事件数（0 则跳过 HTTP）。 */
  picked: number;
  /** ov 受理的事件 ID 清单（含幂等重复；批内顺序即受理顺序）。 */
  acceptedEventIds: string[];
  /** ov 实际应用（新版本生效）的条数。 */
  applied: number;
}

const DEFAULT_BATCH_SIZE = 100;
const DEFAULT_TIMEOUT_MS = 30_000;

/** ov 侧 content-batches 端点路径（全局前缀 v1 + InternalSyncGuard 域）。 */
const CONTENT_BATCHES_PATH = '/v1/internal/sync/content-batches';

/** deliverPendingOnce 的排空轮数上限（每轮一批 ≤ batchSize；防御性熔断）。 */
const DELIVER_MAX_ROUNDS = 100;

/**
 * 一次性排空 outbox（P2 Task 6）：循环 runOnce 直至无未投递事件。
 * 消费方：viewer e2e 夹具（flushSync，种子后经真实管线同步到 ov）与运维手工
 * 触发——与常驻 main 循环同一 OutboxSender，无旁路语义；与常驻发送器并发
 * 亦安全（eventId 幂等 + deliveredAt 条件回写，两侧竞争不双发不丢发）。
 */
export async function deliverPendingOnce(
  options: OutboxSenderOptions,
): Promise<void> {
  const sender = new OutboxSender(options);
  for (let round = 0; round < DELIVER_MAX_ROUNDS; round++) {
    const stats = await sender.runOnce();
    if (stats.picked === 0) {
      return;
    }
  }
  throw new Error(
    `deliverPendingOnce 超时：${DELIVER_MAX_ROUNDS} 轮后 outbox 仍有未投递事件`,
  );
}

export class OutboxSender {
  private readonly batchSize: number;
  private readonly timeoutMs: number;
  private readonly endpoint: string;

  constructor(private readonly options: OutboxSenderOptions) {
    this.batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE;
    this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
    // 基址容错：吞并尾斜杠再拼路径，端点只此一处。
    this.endpoint = `${options.ovApiUrl.replace(/\/+$/, '')}${CONTENT_BATCHES_PATH}`;
  }

  /** 单轮推送：取批 → 线格式 → 签名推送 → 按受理回写 deliveredAt。 */
  async runOnce(): Promise<PushStats> {
    const rows = await this.pickBatch();
    if (rows.length === 0) {
      return { picked: 0, acceptedEventIds: [], applied: 0 };
    }
    const events = rows.map((row) => this.toWireEvent(row));
    const response = await this.push(events);
    // 只标记 ov 明确受理的 eventId（且仍为未投递），幂等可重入。
    if (response.acceptedEventIds.length > 0) {
      await this.options.prisma.outbox_events.updateMany({
        where: {
          id: { in: response.acceptedEventIds },
          deliveredAt: null,
        },
        data: { deliveredAt: new Date() },
      });
    }
    this.options.logger?.debug(
      {
        picked: rows.length,
        accepted: response.acceptedEventIds.length,
        applied: response.applied,
      },
      'outbox batch pushed',
    );
    return {
      picked: rows.length,
      acceptedEventIds: response.acceptedEventIds,
      applied: response.applied,
    };
  }

  /**
   * 两段取数（见文件头排序说明）：high 段占满批则不发第二查；两查之间存在
   * 微小时序窗口（新事件落入两查之间）——对轮询器无害，下一轮补上。
   */
  private async pickBatch(): Promise<OutboxRow[]> {
    const prisma = this.options.prisma;
    const high = await prisma.outbox_events.findMany({
      where: { deliveredAt: null, priority: 'high' },
      orderBy: [{ occurredAt: 'asc' }, { id: 'asc' }],
      take: this.batchSize,
      // 只取发送器关心的列（payload JsonValue 原样透传）。
      select: {
        id: true,
        type: true,
        albumId: true,
        photoId: true,
        entityVersion: true,
        priority: true,
        payload: true,
        occurredAt: true,
      },
    });
    if (high.length >= this.batchSize) {
      return high;
    }
    const rest = await prisma.outbox_events.findMany({
      where: { deliveredAt: null, priority: { not: 'high' } },
      orderBy: [{ occurredAt: 'asc' }, { id: 'asc' }],
      take: this.batchSize - high.length,
      select: {
        id: true,
        type: true,
        albumId: true,
        photoId: true,
        entityVersion: true,
        priority: true,
        payload: true,
        occurredAt: true,
      },
    });
    return [...high, ...rest];
  }

  /**
   * outbox 行 → SyncEvent 线格式。entityId 按实体类型取自不同位置：
   * album → albumId 列；photo → photoId 列；category/asset → 载荷内的
   * categoryId/assetId（outbox 表无此二列）。映射缺料抛错（fail-fast：
   * 本轮不推、不标记，错误日志暴露；cn 侧发射器按构造保证不缺料）。
   */
  private toWireEvent(row: OutboxRow): SyncEventWire {
    const type = row.type as OutboxEvent['type'];
    const entityType = syncEntityTypeOfOutboxType(type);
    const payload = (row.payload ?? {}) as Record<string, unknown>;
    let entityId: string;
    if (entityType === 'album') {
      entityId = row.albumId;
    } else if (entityType === 'photo') {
      if (row.photoId === null) {
        throw new Error(`outbox 事件 ${row.id}（${type}）缺 photoId，无法推导 entityId`);
      }
      entityId = row.photoId;
    } else if (typeof payload.categoryId === 'string') {
      entityId = payload.categoryId;
    } else if (typeof payload.assetId === 'string') {
      entityId = payload.assetId;
    } else {
      throw new Error(`outbox 事件 ${row.id}（${type}）载荷缺实体 ID，无法推导 entityId`);
    }
    const tombstone = isTombstoneOutboxType(type);
    return {
      eventId: row.id,
      schemaVersion: 1,
      entityType,
      entityId,
      albumId: row.albumId,
      entityVersion: row.entityVersion,
      // 线格式 tombstone 显式携带（生成类型必填：openapi-typescript 把带
      // default 的属性渲染为非可选）。
      tombstone,
      priority: row.priority === 'high' ? 'high' : 'normal',
      payload: payload as SyncEventWire['payload'],
    };
  }

  /**
   * 签名推送：同一 JSON.stringify 产物既作签名输入也作请求体（原始字节一致，
   * 接收端 rawBody 验签才能通过——见 contracts/internal-sync.ts 头注）。
   * 非 2xx / 响应形状异常均抛错（调用方整批重试）。
   */
  private async push(events: SyncEventWire[]): Promise<ContentBatchResponse> {
    const rawBody = JSON.stringify({ batch: events });
    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(
        `content-batches 请求失败（${this.endpoint}）：${err instanceof Error ? err.message : String(err)}`,
      );
    }
    if (!response.ok) {
      const body = await response.text().catch(() => '');
      throw new Error(
        `content-batches 非 2xx（${this.endpoint}）：status=${response.status} body=${body.slice(0, 500)}`,
      );
    }
    const data: unknown = await response.json();
    if (
      typeof data !== 'object' ||
      data === null ||
      !Array.isArray((data as ContentBatchResponse).acceptedEventIds) ||
      !((data as ContentBatchResponse).acceptedEventIds as unknown[]).every(
        (id) => typeof id === 'string',
      ) ||
      typeof (data as ContentBatchResponse).applied !== 'number'
    ) {
      throw new Error('content-batches 响应形状异常（acceptedEventIds/applied 缺失）');
    }
    return data as ContentBatchResponse;
  }
}