Merge branch 'feat/friday-ingestion' into test

apps/pac-docs/content/docs/integration/friday-push-payload.mdx

@@ -245,7 +245,67 @@ icon: FileJson
 
 ---
 
-## 7. 语义澄清记录(2026-07,源码 + 数据双证)
+## 7. 对接自测(预检 + 推送记录)
+
+对接期不必盲推 —— PAC 提供两个自助工具,**测试环境** `https://pac.jarvismedical.asia`:
+
+### 7.1 预检:`?dryRun=1`(强烈建议先跑)
+
+```
+POST /pac/v1/push/rows?dryRun=1
+```
+跑**完整管线**(transforms + 装配 + 归一 + 租户解析)但**不落任何库、不触发重算**,响应告诉你"若正式推会发生什么",并带**样本 canonical** —— 直接核对 PAC 把你的原生行翻译成了什么:
+
+```jsonc
+{"code":0,"data":{
+  "dryRun": true,
+  "accepted": 6,               // 收到 6 行
+  "transactionsWritten": 5,    // 其中 5 行会落库(差值 = 被 PAC 规则过滤,如噪音 status)
+  "failed": 0,
+  "mappingMisses": 0,          // ⚠️ >0:有原值没映射上(落 _default)
+  "suspectFields": 0,          // ⚠️ >0:疑似列改名/缺列
+  "samples": [ { "resource": "payment", "canonical": { "externalId": "…", "amount": 40000, … } } ]
+}}
+```
+
+**自测通过标准**:`failed=0` 且 `mappingMisses=0` 且 `suspectFields=0`,再看 `samples` 里的字段值对不对 —— 都 OK 才去掉 `?dryRun=1` 正式推。
+
+### 7.2 推送记录:事后可查(两种查法)
+
+**① 免登录 —— 用已有 push 凭据验签(联调推荐)**
+```
+GET /pac/v1/push/logs?limit=50
+Headers: X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature
+```
+鉴权方式与推送完全一致,**不必为联调另开账号**。GET 无 body,签名对**空串**计算:
+`hex(HMAC-SHA256("<timestamp>.", secret))`(注意 timestamp 后那个点)。
+这**不是公开接口** —— 没有正确签名拿不到任何数据。
+
+**② 登录态 —— 宿主自助页同源接口**
+```
+GET /pac/v1/admin/host/self/push-logs?limit=50
+```
+
+两者返回**同一份数据**:按时间倒序最近 N 批(默认 50、上限 200),每批含
+`source / status / dryRun / fetched / transactionsWritten / duplicates / failed / errorMessage`。
+当时的同步响应没留存也能事后查:**哪批错了、错在哪**。预检批次由 `dryRun:true` 标识,不与正式推送混淆。
+
+### 7.3 其他可用面
+
+| 用途 | 地址 |
+|---|---|
+| 交互式 API 文档(可试调) | `/api/docs` |
+| 本契约文档 | `/docs/integration/friday-push-payload` |
+| 宿主自助页(患者数 / 24h 交易 / 上次 push / 失败率) | `/admin/host` |
+| 队列面板(看 push 触发的画像重算) | `/admin/queues` |
+
+> **常见错误码**:`10106` 缺 header / 签名不匹配(或时间戳偏差 >5min)· `10001` source 拼错
+> (响应会列出可选源表名)· `10002` body 校验失败 · `10003` 并发超限(退避重发即可,幂等安全)
+> · `30802` 整批失败,疑似关键字段缺名/改名。
+
+---
+
+## 8. 语义澄清记录(2026-07,源码 + 数据双证)
 
 以下曾是开放问题,现已定案 —— 推送方无需再确认,列此备查:
 

apps/pac-service/src/modules/admin/dto/admin.dto.ts

@@ -47,3 +47,21 @@ const ColdImportStatusSchema = z.object({
   progress: z.object({ done: z.number(), total: z.number() }).nullable(),
 });
 export class ColdImportStatusDto extends createZodDto(ColdImportStatusSchema) {}
+
+/// 宿主推送记录(sync_logs direction=push 的对宿主视图)—— 对接期自助排查:
+/// 哪一批、什么时候、收了几行、落了几条、去重几条、失败几条、错在哪。
+const PushLogSchema = z.object({
+  id: z.string(),
+  source: z.string().describe('推送的源表名'),
+  status: z.string().describe('success / partial / failed'),
+  dryRun: z.boolean().describe('true = 预检批次(未落库)'),
+  fetched: z.number().int().describe('收到的行数'),
+  transactionsWritten: z.number().int().describe('实际落库事件数(与 fetched 的差 = 被 PAC 规则过滤)'),
+  duplicates: z.number().int().describe('幂等命中数(重推安全的证据)'),
+  failed: z.number().int(),
+  errorMessage: z.string().nullable(),
+  startedAt: z.string(),
+  endedAt: z.string().nullable(),
+});
+const PushLogsSchema = z.object({ items: z.array(PushLogSchema) });
+export class PushLogsDto extends createZodDto(PushLogsSchema) {}

apps/pac-service/src/modules/admin/host-self.controller.ts

@@ -6,6 +6,7 @@ import {
   Param,
   Patch,
   Post,
+  Query,
   UploadedFile,
   UseInterceptors,
 } from '@nestjs/common';
@@ -26,6 +27,7 @@ import {
   HostDetailDto,
   HostStatsDto,
   HostSummaryDto,
+  PushLogsDto,
   RemovePushSecretResponseDto,
   RotateCallbackSecretResponseDto,
   RotatePushSecretResponseDto,
@@ -78,6 +80,20 @@ export class HostSelfAdminController {
     return this.hosts.getStats(scope.hostId);
   }
 
+  @Get('push-logs')
+  @ZodResponse({ status: 200, type: PushLogsDto })
+  @ApiOperation({
+    summary: '当前宿主的推送记录(对接自助排查)',
+    description:
+      '按时间倒序列最近 N 批 push(默认 50,上限 200)。每批含 source / 收行数 / 落库数 / ' +
+      '去重数 / 失败数 / 报错原因;dryRun=true 的是 ?dryRun=1 预检批次。' +
+      '用途:当时的同步响应没留存也能事后查——对接期定位"哪批错了、错在哪"。',
+  })
+  getPushLogs(@TenantScope() scope: TenantScopeContext, @Query('limit') limit?: string) {
+    const n = Number.parseInt(limit ?? '', 10);
+    return this.hosts.getPushLogs(scope.hostId, Number.isFinite(n) && n > 0 ? n : 50);
+  }
+
   @Patch()
   @ZodResponse({ status: 200, type: HostSummaryDto })
   @ApiOperation({

apps/pac-service/src/modules/admin/hosts.service.ts

@@ -17,6 +17,7 @@ import type {
   PullConfig,
   ActionUrlsConfig,
 } from '@pac/types';
+import { SyncDirection } from '@pac/types';
 import { PrismaService } from '../../prisma/prisma.service';
 
 /**
@@ -300,6 +301,37 @@ export class HostsService {
     };
   }
 
+  /**
+   * 宿主推送记录(自助排查)—— 列最近 N 批 push 的结果。
+   *
+   * 对接期宿主开发最需要的"事后可查":当时的同步响应若没打进日志,这里仍能看到
+   * 每批 source / 收行数 / 落库数 / 去重 / 失败 / 报错原因。`?dryRun=1` 的预检批次
+   * 由 triggeredBy 前缀 `push-dryrun:` 标识,单独打标,不与正式推送混淆。
+   */
+  async getPushLogs(hostId: string, limit = 50) {
+    const take = Math.min(Math.max(limit, 1), 200);
+    const rows = await this.prisma.syncLog.findMany({
+      where: { hostId, direction: SyncDirection.PUSH },
+      orderBy: { startedAt: 'desc' },
+      take,
+    });
+    return {
+      items: rows.map((r) => ({
+        id: r.id,
+        source: r.resource,
+        status: r.status,
+        dryRun: (r.triggeredBy ?? '').startsWith('push-dryrun:'),
+        fetched: r.fetched ?? 0,
+        transactionsWritten: r.transactionsWritten ?? 0,
+        duplicates: r.duplicates ?? 0,
+        failed: r.failed ?? 0,
+        errorMessage: r.errorMessage,
+        startedAt: r.startedAt.toISOString(),
+        endedAt: r.endedAt?.toISOString() ?? null,
+      })),
+    };
+  }
+
   // ─────────────────────────────────────────────────────────
   // Remove push secret(按 suffix 定位,精确移除某个 grace key)
   // ─────────────────────────────────────────────────────────

apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts

@@ -286,6 +286,9 @@ export class ColdImportService {
     source: string; // 原生源表名(= manifest tables[].table / transform 终端表)
     rows: Record<string, unknown>[];
     triggeredBy?: string;
+    /// 预检:跑完整管线(transforms + 装配 + 归一 + tenant 解析)但**不落任何库**,
+    /// 用于宿主对接自测——看 accepted/失败数与样本 canonical,确认 PAC 理解对了再正式推。
+    dryRun?: boolean;
   }): Promise<{
     syncLogId: string;
     perResource: PerResourceStats[];
@@ -333,7 +336,9 @@ export class ColdImportService {
         tenantId: 't_pending', // push 多 tenant 由行决定;finalize 时回填首个 seen tenant
         direction: SyncDirection.PUSH,
         resource: opts.source,
-        triggeredBy: opts.triggeredBy ?? `push:${opts.hostName}:${opts.source}`,
+        triggeredBy:
+          opts.triggeredBy ??
+          `${opts.dryRun ? 'push-dryrun' : 'push'}:${opts.hostName}:${opts.source}`,
         status: SyncStatus.RUNNING,
       },
     });
@@ -362,19 +367,19 @@ export class ColdImportService {
         if (cfg.canonical === 'patient') {
           perResource.push(
             await this.processPatients(
-              transformed, cfg, host.id, tenantResolver, seenTenants, normalize, false, true,
+              transformed, cfg, host.id, tenantResolver, seenTenants, normalize, !!opts.dryRun, true,
             ),
           );
         } else if (cfg.canonical === 'patient_relation') {
           perResource.push(
             await this.processPatientRelations(
-              transformed, cfg, host.id, tenantResolver, normalize, false,
+              transformed, cfg, host.id, tenantResolver, normalize, !!opts.dryRun,
             ),
           );
         } else if (cfg.canonical === 'patient_return_visit') {
           perResource.push(
             await this.processPatientReturnVisits(
-              transformed, cfg, host.id, tenantResolver, normalize, false,
+              transformed, cfg, host.id, tenantResolver, normalize, !!opts.dryRun,
             ),
           );
         }
@@ -388,7 +393,7 @@ export class ColdImportService {
           seenTenants,
           normalize,
           syncLog.id,
-          false,
+          !!opts.dryRun,
         );
         perResource.push(stats);
       }

apps/pac-service/src/modules/sync/push/push-receiver.service.ts

@@ -49,14 +49,18 @@ export class PushReceiverService {
     hostName: string;
     source: string;
     rows: Record<string, unknown>[];
+    /// 预检(?dryRun=1):跑完整管线但不落库,也不触发重算。宿主对接自测用。
+    dryRun?: boolean;
   }): Promise<PushRowsResponse> {
     const { hostId, hostName, source, rows } = input;
+    const dryRun = !!input.dryRun;
 
     const r = await this.coldImport.ingestRawTables({
       hostName,
       source,
       rows,
-      triggeredBy: `push:${hostName}:${source}`,
+      dryRun,
+      triggeredBy: `${dryRun ? 'push-dryrun' : 'push'}:${hostName}:${source}`,
     });
 
     const agg = r.perResource.reduce(
@@ -81,8 +85,9 @@ export class PushReceiverService {
 
     // 事件驱动:touched 患者入队 persona-recompute(去抖:存量批次重放时窗内合并,见 QueueProducer)
     // ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。
+    // 预检不产生任何副作用 → 不入队(touched 本就为空,此处显式短路,语义更清楚)。
     let personaEnqueued = 0;
-    for (const { patientId, tenantId } of r.touched) {
+    for (const { patientId, tenantId } of dryRun ? [] : r.touched) {
       try {
         await this.queue.enqueuePersonaRecompute(
           { hostId, tenantId, patientId, triggeredBy: `push:${hostName}` },
@@ -101,6 +106,13 @@ export class PushReceiverService {
         `txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} personaEnqueued=${personaEnqueued}`,
     );
 
+    // 预检:回样本 canonical(各资源首行),供宿主核对 PAC 把原生行翻译成了什么
+    const samples = dryRun
+      ? r.perResource
+          .filter((s) => (s.sampleCanonical?.length ?? 0) > 0)
+          .map((s) => ({ resource: s.resource, canonical: s.sampleCanonical[0] }))
+      : [];
+
     return {
       syncLogId: r.syncLogId,
       source,
@@ -111,6 +123,8 @@ export class PushReceiverService {
       personaEnqueued,
       mappingMisses: r.mappingMisses.length,
       suspectFields: r.suspectFields.length,
+      dryRun,
+      samples,
     };
   }
 

apps/pac-service/src/modules/sync/push/push.controller.ts

 import {
   Body,
   Controller,
+  Get,
   Headers,
   HttpCode,
   Post,
+  Query,
   Req,
 } from '@nestjs/common';
 import type { Request } from 'express';
-import { ApiOperation, ApiTags } from '@nestjs/swagger';
+import { ApiOperation, ApiQuery, ApiTags } from '@nestjs/swagger';
 import { ApiCode } from '@pac/types';
 import { BizError } from '../../../common/errors/biz-error';
 import { Public } from '../../../common/decorators/public.decorator';
 import { HmacVerifier } from './hmac-verifier.service';
 import { PushReceiverService } from './push-receiver.service';
+import { HostsService } from '../../admin/hosts.service';
 import {
   PushEventsRequestSchema,
   type PushEventsRequest,
@@ -44,6 +47,7 @@ export class PushController {
   constructor(
     private readonly verifier: HmacVerifier,
     private readonly receiver: PushReceiverService,
+    private readonly hosts: HostsService,
   ) {}
 
   /**
@@ -137,14 +141,23 @@ export class PushController {
   @Public()
   @HttpCode(200)
   @ApiOperation({
-    summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签)',
-    description: 'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。Body: { tenantId?, source, rows[] }。',
+    summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签;?dryRun=1 预检不落库)',
+    description:
+      'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。Body: { source, rows[] }。' +
+      ' 加 ?dryRun=1 走**预检**:跑完整管线(transforms+装配+归一)但不落库、不触发重算,' +
+      '响应带各资源样本 canonical —— 对接自测先预检到 failed=0 且 mappingMisses/suspectFields 为 0,再正式推。',
+  })
+  @ApiQuery({
+    name: 'dryRun',
+    required: false,
+    description: "预检开关:'1' / 'true' / 'yes' = 跑完整管线但不落库、不触发重算,响应带样本 canonical",
   })
   async receiveRows(
     @Req() req: Request & { rawBody?: Buffer },
     @Headers('x-pac-host-id') hostIdHeader: string | undefined,
     @Headers('x-pac-timestamp') timestampHeader: string | undefined,
     @Headers('x-pac-signature') signatureHeader: string | undefined,
+    @Query('dryRun') dryRunQuery: string | undefined,
     @Body() body: PushRowsRequest,
   ) {
     const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(body);
@@ -158,13 +171,53 @@ export class PushController {
 
     const parsed = this.parseOrReject(() => PushRowsRequestSchema.parse(body));
 
+    // ?dryRun=1 / true / yes 均视为预检(宿主客户端拼 query 习惯不一)
+    const dryRun = ['1', 'true', 'yes'].includes((dryRunQuery ?? '').trim().toLowerCase());
+
     return this.withHostLock(hostId, hostName, () =>
       this.receiver.receiveRows({
         hostId,
         hostName,
         source: parsed.source,
         rows: parsed.rows,
+        dryRun,
       }),
     );
   }
+
+  /**
+   * 推送记录查询 —— **用已有 push 凭据验签,无需另外登录**(对接自测友好)。
+   *
+   * 与登录态的 `GET /admin/host/self/push-logs` 返回同一份数据(同一实现),区别只在鉴权方式:
+   * 这里复用宿主推送时已经在用的 HMAC 三件套,故不必为联调再开账号;
+   * 也**不是公开接口** —— 没有正确签名拿不到任何数据,不会泄露宿主推送情况。
+   *
+   * GET 无 body,签名对**空串**计算:`hex(HMAC-SHA256(`${timestamp}.`, secret))`。
+   */
+  @Get('logs')
+  @Public()
+  @HttpCode(200)
+  @ApiOperation({
+    summary: '查询本宿主推送记录(HMAC 验签,免登录)',
+    description:
+      'Headers 同 push:X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。' +
+      'GET 无 body,签名对空串算:hex(HMAC-SHA256(`<timestamp>.`, secret))。' +
+      '返回最近 N 批(默认 50、上限 200):source / status / dryRun / 收行数 / 落库数 / 去重 / 失败 / 报错。',
+  })
+  @ApiQuery({ name: 'limit', required: false, description: '返回条数,默认 50、上限 200' })
+  async logs(
+    @Headers('x-pac-host-id') hostIdHeader: string | undefined,
+    @Headers('x-pac-timestamp') timestampHeader: string | undefined,
+    @Headers('x-pac-signature') signatureHeader: string | undefined,
+    @Query('limit') limit?: string,
+  ) {
+    const { hostId } = await this.verifier.verify({
+      hostIdHeader,
+      timestampHeader,
+      signatureHeader,
+      rawBody: '', // GET 无 body
+    });
+    const n = Number.parseInt(limit ?? '', 10);
+    return this.hosts.getPushLogs(hostId, Number.isFinite(n) && n > 0 ? n : 50);
+  }
 }

apps/pac-service/src/modules/sync/push/push.schema.ts

@@ -87,9 +87,15 @@ export const PushRowsResponseSchema = z.object({
   duplicates: z.number().int(),
   failed: z.number().int(),
   personaEnqueued: z.number().int().describe('入队画像重算的患者数(plan 不在此触发,由定时任务保)'),
-  /// 宿主自检信号(代替 cold-import 的 dry-run):样本批推完看这两个数,>0 先停下联系 PAC。
+  /// 宿主自检信号:样本批推完看这两个数,>0 先停下联系 PAC。
   mappingMisses: z.number().int().describe('映射覆盖缺口种类数(有原值落 _default)'),
   suspectFields: z.number().int().describe('疑似宿主改表项数(列缺席/新增,vs 历史基线)'),
+  /// 预检模式(?dryRun=1):跑完整管线(transforms+装配+归一+tenant 解析)但**不落任何库**,
+  /// 数字表示"若正式推会发生什么"。对接自测用:先预检到 failed=0 且两个自检信号为 0,再正式推。
+  dryRun: z.boolean().describe('true = 本次为预检,未落库'),
+  samples: z
+    .array(z.object({ resource: z.string(), canonical: z.unknown() }))
+    .describe('预检样本:各 canonical 资源首行的翻译结果(仅 dryRun 返回,供宿主核对 PAC 是否理解正确)'),
 });
 
 export type PushRowsResponse = z.infer<typeof PushRowsResponseSchema>;

apps/pac-service/src/modules/sync/sync.module.ts

@@ -32,8 +32,11 @@ import { HmacVerifier } from './push/hmac-verifier.service';
  *     - AssemblerModule(raw → canonical;真实 HTTP adapter 才需要;mock/push 跳过)
  *     - PipelineModule(synthesizer + parser pipeline)
  */
+import { AdminModule } from '../admin/admin.module';
+
 @Module({
   imports: [
+    AdminModule, // push/logs 复用 HostsService.getPushLogs(与登录态自助页同一实现)
     FactsModule,
     AssemblerModule,
     TransformsModule,