Skip to content

P
pac
Commit 01208245 by luoqi
Options

docs(pac): align push receipts and failed-row handling

parent 3a057ec5
Pipeline #3765 failed in 0 seconds
Showing with 90 additions and 31 deletions
apps/pac-docs/content/docs/architecture/data-model.mdx
......@@ -142,7 +142,9 @@ patient_facts.transactionIds → patient_transactions.id → rawPayload
**`agent_invocations`** — AI 调用审计账本(append-only,跨所有层共享)。`kind` / `call_key` / `prompt_version` / `model_provider` / `model_name` / `workflow_run_id` / `input_snapshot` / `output` / token 明细 / `cost_yuan` / `status`。每次 LLM 调用两段式落账(start=running → end 回填),是 audit / replay / 成本核算唯一真理源。AI 解析结果归 `output`,**不回写 fact**。
**`sync_logs`** — 同步运行账本。`direction`(pull/push)/ `resource` / `status`(running/success/partial/failed)/ `cursor_before`/`after`(断点续传)/ `triggered_by`(含 `reconcile:<date>` 对账)/ 计数恒等式 `fetched = transactions_written + duplicates + failed` + `facts_emitted` / `metadata`(cohort 进度 + 漂移审计 `suspectFields`)。
**`sync_logs`** — 同步运行账本。`direction`(pull/push)/ `resource` / `status`(running/success/partial/failed)/ `cursor_before`/`after`(pull 游标;push 留空)/ `triggered_by`(含 `reconcile:<date>` 对账)/ `fetched`、`transactions_written`、`duplicates`、`failed`、`facts_emitted` / `metadata`(cohort 进度 + 漂移审计 `suspectFields`)。
Push 的 `fetched` 是收到的原生行数;主档 / 关系 / 回访直接 upsert 不计入流水数,原生行还可能被拆分或过滤,各计数不保证相加等于 `fetched`。有成功 upsert、新流水或幂等命中且部分失败时记 `partial`;入库前拒收行不另存原文或自动隔离。详见 [Push 回执](/docs/integration/channel-push#9-回执与重试)。
---
......
apps/pac-docs/content/docs/ingestion.mdx
......@@ -137,6 +137,12 @@ pnpm sync -- --dir=./data/<host> --full # 忽略游标强制全量(灾
## 三、数据完整性
**入站失败与原文留存**:
- 校验失败的记录拒收,有效记录继续写入;Push 的批次分类与重试见 [Push 回执](/docs/integration/channel-push#9-回执与重试)。
- 主档、关系、回访直接 upsert,不生成原始流水;入库前拒收行仅有失败计数和错误日志,不保存完整原文、不进入自动隔离 / 重试队列。
- 已写入的 `patient_transactions` 保留 `rawPayload`;事实校验拒收用 `factsFailed` / `factRejects` 报告。宿主保留源事件及 payload,负责修正重推或持久化隔离。
**幂等(数据库双 UNIQUE)** —— 任何并发 / 重跑结果一致:
- `patient_transactions`:partial UNIQUE `(host_id, tenant_id, source_event_id) WHERE source_event_id NOT NULL`
......
apps/pac-docs/content/docs/integration/channel-push.mdx
......@@ -16,7 +16,7 @@ icon: Send
| 需要了解 | 自己的数据字典 | PAC 的字段与枚举契约(§6) |
| 字段映射 | PAC 完成 | 宿主按契约组装,码值由 PAC 映射 |
| 一次性成本 | 提供一份数据字典 | 按契约开发组装逻辑 |
| 后续变更 | 映射调整在 PAC 侧完成,无需重发数据 | 契约变更需双方同步 |
| 后续变更 | 已保存原始流水可由 PAC 重解析;未入库行修正后重推 | 契约变更需双方同步 |
多数系统选 A。鉴权、幂等、推送顺序等规则两形态一致。
......@@ -219,7 +219,7 @@ const sig = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
| 场景 | `code` | 处理 |
| ------------------ | ------- | ----------------------------------------------------------------------- |
| 成功 | `0` | — |
| 请求处理完成(含部分成功) | `0` | 继续检查 `failed` / `factsFailed`,不能仅凭 `code=0` 推进消费位点 |
| 验签失败 | `10106` | 检查 secret、时钟、签名串 |
| 字段校验失败 / 批量超限(条数或字节) | `10002` | 修正后重发(字节超限见 §10,减小单批行数) |
| source 表名不识别 | `10001` | 核对表名 |
......@@ -230,15 +230,49 @@ const sig = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
未收到响应时直接重试,幂等保证安全。
形态 A 成功响应中的自检字段:
**形态 A 的逐行处理**:
| 批次结果 | 回执 | 推送记录 `status` |
|---|---|---|
| 无逐行失败 | `code=0, failed=0` | `success` |
| 有成功 upsert、新流水或幂等命中,同时有失败 | `code=0, failed>0` | `partial` |
| 有失败,且没有任何上述进展 | `code=30802` | `failed` |
有效记录照常写入;校验不通过的记录拒收,不补猜患者身份。主档、患者关系、回访直接 upsert,
成功时也可能 `transactionsWritten=0, duplicates=0`,不能据此判定整批失败。
形态 A 正常回执中的自检字段:
| 字段 | 含义 | 大于 0 时 |
|---|---|---|
| `failed` | 本批**没进 PAC** 的行数 | 零星可忽略,持续升高联系 PAC |
| `factsFailed` | 行进了、但**派生不出可用事实**的条数 | 看 `factRejects` 定位到自己源库那一行 |
| `accepted` | 收到的原生行数,不是成功写入数 | 不作为消费完成判据 |
| `transactionsWritten` | 新增流水数;不含主档、关系、回访的直接 upsert | 不与 `accepted` 作成功率等式 |
| `failed` | 装配 / 写入失败计数,按派生资源累计 | 修正后重推,或由宿主持久化隔离后处理;不能直接忽略 |
| `factsFailed` | 流水已保存,但部分派生事实被校验拒收的数量 | 看 `factRejects` 定位与修正 |
| `mappingMisses` | 存在未映射的原值 | 暂停推送,发样本给 PAC |
| `suspectFields` | 与历史相比列发生增减 | 确认表结构是否变更,同步 PAC |
一条原生行可能拆成多条流水或被业务规则过滤,上述计数不保证相加等于 `accepted`。
`status` 在推送记录中查询,不是 `/push/rows` 回执中的字段。
例如 200 条关系中 197 条有效、3 条缺少身份字段,回执节选为:
```json
{
"code": 0,
"data": {
"source": "customer_referee_circle",
"accepted": 200,
"transactionsWritten": 0,
"duplicates": 0,
"failed": 3,
"factsFailed": 0
}
}
```
该批推送记录为 `partial`;197 条有效关系已写入,3 条失败关系未写入。
`factsFailed > 0` 时,`factRejects` 给出定位明细(最多 10 条,同一种错通常整批同因):
```json
......@@ -261,6 +295,13 @@ const sig = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
出于隐私,`factRejects` 只给定位信息与字段级原因,**不回传记录原文**。
**失败留档与重试责任**:
- PAC 当前记录失败计数和错误日志,不保存入库前被拒行的完整原文,不提供这些行的逐条主键清单,也没有自动隔离 / 重试队列。
- 主档、关系、回访仅保存当前态;这些直接 upsert 资源不生成原始流水。只有已写入的 `patient_transactions` 保留 `rawPayload`。
- 宿主保留 outbox / 事件日志及原始 payload。数据错误修正后重推;按业务策略隔离失败事件时,先持久化并留审计,再决定是否推进位点。
- 同一批重推具有幂等性,但不修正原数据只会重复同一错误。统一位点的消费端若遇到 `failed>0` 就整批重试,后续事件会被该批阻挡。
### 9.2 补漏
每晚将最近 N 天数据全量重推一遍:已推的跳过、漏推的补上、变更的升版本。
......@@ -302,7 +343,7 @@ PAC 侧对推送中断有告警。
- [ ] 组织三 id 正确(§3)
- [ ] 删除走软删(§7.4)
- [ ] 本地 outbox 持久化 + 异步 worker(业务写库与入队同事务)
- [ ] 按 `code` 处理响应(§9.1),`10003` 退避重发
- [ ] 按 `code`、`failed`、`factsFailed` 处理响应(§9.1);失败事件修正 / 持久化隔离后才推进位点,`10003` 退避重发
- [ ] 存量:样本批验证后分批重放(§8)
- [ ] 每晚幂等补推最近 N 天(§9.2)
......
apps/pac-docs/content/docs/integration/friday-push-payload.mdx
......@@ -323,8 +323,8 @@ icon: FileJson
| `return_visit_method` | string/number | | 回访方式——`1`打电话 `2`发微信 `3`发短信 |
| `created_gmt_at` / `updated_gmt_at` | string(datetime) | ✅ | 创建 / 末次修改时间 |
**建议一并 inline 的 `customer_task` 列**(PAC 当前不映射,原样存进 `raw_payload` 备用,
将来要用不必再找宿主改):`suggested_return_person`(建议回访人)、`actual_return_person` /
**可一并 inline 的 `customer_task` 列**(PAC 当前不映射、不保存这些额外原文;
需要使用时另行补映射并重推):`suggested_return_person`(建议回访人)、`actual_return_person` /
`actual_return_time`(实际回访人/时间)、`customer_status`(客户状态)、`is_first`(初诊/复诊)。
> **`task_director` 是「该任务当时派给谁」,不是患者的专属客服。**
......@@ -345,13 +345,13 @@ icon: FileJson
```
POST /pac/v1/push/rows?dryRun=1
```
跑**完整管线**(transforms + 装配 + 归一 + 租户解析)但**不落任何库、不触发重算**,响应告诉你"若正式推会发生什么",并带**样本 canonical** —— 直接核对 PAC 把你的原生行翻译成了什么:
跑 transforms + 装配 + 归一 + 租户解析,**不写业务数据、不触发重算**,保留预检推送记录。响应返回预期计数与**样本 canonical**;不执行实际写库与事实派生校验,正式推送仍需检查回执:
```jsonc
{"code":0,"data":{
"dryRun": true,
"accepted": 6, // 收到 6 行
"transactionsWritten": 5, // 其中 5 行会落库(差值 = 被 PAC 规则过滤,如噪音 status)
"transactionsWritten": 5, // 预计生成 5 条流水;不含直接 upsert,不等于成功原生行数
"failed": 0,
"mappingMisses": 0, // ⚠️ >0:有原值没映射上(落 _default)
"suspectFields": 0, // ⚠️ >0:疑似列改名/缺列
......@@ -379,7 +379,7 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
两者返回**同一份数据**:按时间倒序最近 N 批(默认 50、上限 200),每批含
`source / status / dryRun / fetched / transactionsWritten / duplicates / failed / errorMessage`。
当时的同步响应没留存也能事后查:**哪批错了、错在哪**。预检批次由 `dryRun:true` 标识,不与正式推送混淆。
可事后查询批次状态与失败计数;`errorMessage` 是批次级错误,逐行校验原因看 PAC 日志。当前不返回失败行主键或原文。预检批次由 `dryRun:true` 标识,不与正式推送混淆。
### 8.3 其他可用面
......@@ -405,7 +405,7 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
| 时间口径 | MySQL `datetime` = 北京墙钟(`_gmt_` 命名系惯例误导);**Mongo `Date` 是北京墙钟伪装成 UTC** | `clinicTime`/`createdGmtAt` 小时分布呈营业双峰;按真 UTC 解释则半数病历落深夜 |
| 金额单位 | 元(`decimal(9,2)`) | 洁治单价均值 ¥310、已结算客单均值 ¥1,730 |
| `relationship`(默认号归属) | `PhoneRelationshipEnum` 1-9(见 §2;宿主挑默认号后 inline) | customer 服务枚举类 |
| `referee_relationship` | `RecommendRelationshipEnum` 17 码,**语义 = 本行 customer 是 referee 的 X**(PAC 侧按逆关系映射) | 枚举类 `reverseValue()` + 双方年龄差数据 |
| `referee_relationship` | `RecommendRelationshipEnum` 17 码,**语义 = 本行 referee 是 customer 的 X**(与字段表一致,PAC 同向映射) | 当前 `patient_relation.yaml` 映射及双方年龄差核查 |
| `med_emr_info.treat` vs `dispose` | `treat` = 本次治疗(→ 结构化治疗事实);`dispose` = 处置叙述(→ 病历自由文本) | EMR 接口 DTO 字段注释 |
| `patient_settlement.status` 全 9 值 | `0`未结算 `1`已结算 `2`只生成uuid `3`含退费行 `4`整单反向冲减 `5`重新结算(废弃历史) `6`医生未提交 `7`流程结束 `8`欠款补缴克隆单 | 实体注释 + `savePatientSettlementQk()` 实现 |
| 为何不收 `7`/`8` | `8` 是原单克隆(`receivable_this` 原样复制、`payment_arrears` 才是本次补缴额),入消费会**双计应收**;`7` 为零金额流程关单 | 同上 |
......@@ -496,3 +496,13 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
}]
}
```
---
## 11. 回执与失败处理
- `code=0` 表示请求处理完成;同时检查 `failed` / `factsFailed`。部分成功记为 `partial`,有效行正常写入,失败行拒收。
- 主档、关系、回访直接 upsert;成功也可能 `transactionsWritten=0, duplicates=0`。`accepted` 仅为收到的原生行数。
- 入库前拒收的行不保存完整原文、不进入 PAC 自动隔离 / 重试队列;宿主保留事件 payload,修正后重推或持久化隔离处理。
- `failed>0` 时不能仅凭 `code=0` 推进位点;不处理坏事件而整批重试会阻挡后续推送。
- 统一口径与示例见 [Push 通道 · 回执与重试](/docs/integration/channel-push#9-回执与重试)。
apps/pac-docs/content/docs/start/what-is-pac.mdx
......@@ -27,7 +27,7 @@ icon: Sparkles
| 维度 | ✅ 确定做 | ❌ 确定不做 |
|---|---|---|
| **数据主权与隔离** | 把宿主数据标准化成一份**不可变的派生副本**，并留存原始 payload 备查；集团（租户）**硬隔离**，互不可见 | 不充当源数据的**权威系统** —— 源数据主权归宿主；不跨集团串数据 |
| **数据主权与隔离** | 标准化宿主数据，已写入的事实流水保留原始 payload 备查；集团（租户）**硬隔离**，互不可见 | 不充当源数据的**权威系统** —— 源数据主权归宿主；不跨集团串数据 |
| **数据范围** | 接收并标准化就诊、治疗、缴费、影像 metadata、病历结构化字段等**宿主侧事实** | 不当**病历系统** —— 完整临床病历（EMR）原文留在宿主 / 医生侧 |
| **事实与产物分层** | 画像、计划、AI 产物归 PAC，**独立存放、全链路可追溯** | 不把自身产物**回写成事实** —— 事实只装宿主侧事实，PAC 产物与事实流严格分层 |
| **智能产出** | 自动派生**患者画像**（价值 / 流失风险 / 沟通偏好，带证据链）；生成**召回清单 + 优先级 + 召回原因 + AI 话术 + 一页摘要**；**客服侧 AI 助手**（工作台 / 企微问答，MCP 工具） | 不**自做影像 AI 诊断** —— 影像 finding 由宿主 / 三方产出，PAC 只摄入并降权使用 |
......
apps/pac-service/prisma/schema.prisma
......@@ -1963,8 +1963,8 @@ model SyncLog {
/// 状态:
/// running — 进行中(用于发现卡死任务)
/// success — 全成功(failed = 0)
/// partial — 部分失败(failed > 0,但有进展,cursor 有推进)
/// failed — 整体失败(连 cursor 都没推进)
/// partial — 有有效进展的部分失败;push 不使用 cursor
/// failed — 无有效进展的整体失败
status String
/// 本次拉取前的游标(push 路径不适用,留 null)
cursorBefore String? @map("cursor_before")
......@@ -1978,11 +1978,11 @@ model SyncLog {
/// 由 parser 衍生写入 patient_facts 的版本数(同一 transaction 可产 0/1/N 个 fact 变化)
factsEmitted Int @default(0) @map("facts_emitted")
/// 幂等去重命中数(source_event_id 已存在,跳过未入库)。
/// 校验恒等式:fetched = transactionsWritten + duplicates + failed
/// 主档/关系/回访直接 upsert、原生行拆分/过滤时,计数不保证相加等于 fetched。
duplicates Int @default(0)
/// 失败条数(zod 校验失败 / parser 异常 / 写库异常 → 转入 failed_transactions quarantine)
/// 装配/写入失败计数;拒收行不另存原文,当前没有 failed_transactions 隔离表。
failed Int @default(0)
/// 失败时的错误信息(单条主因,详细堆栈进应用日志 + quarantine 表)
/// 批次级错误主因;逐行校验原因进应用日志,不提供失败行原文或主键清单。
errorMessage String? @map("error_message") @db.Text
startedAt DateTime @default(now()) @map("started_at") @db.Timestamptz(3)
......
apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts
......@@ -364,8 +364,8 @@ export class ColdImportService {
source: string; // 原生源表名(= manifest tables[].table / transform 终端表)
rows: Record<string, unknown>[];
triggeredBy?: string;
/// 预检:跑完整管线(transforms + 装配 + 归一 + tenant 解析)但**不落任何库**,
/// 用于宿主对接自测——看 accepted/失败数与样本 canonical,确认 PAC 理解对了再正式推。
/// 预检:跑 transforms + 装配 + 归一 + tenant 解析,不写业务数据、不派生事实;
/// 保留预检推送记录,供宿主核对预计计数与样本 canonical。
dryRun?: boolean;
}): Promise<{
syncLogId: string;
......
apps/pac-service/src/modules/sync/push/push-receiver.service.ts
......@@ -49,7 +49,7 @@ export class PushReceiverService {
hostName: string;
source: string;
rows: Record<string, unknown>[];
/// 预检(?dryRun=1):跑完整管线但不落库,也不触发重算。宿主对接自测用。
/// 预检(?dryRun=1):不写业务数据、不派生事实、不触发重算;仍保留预检推送记录。
dryRun?: boolean;
}): Promise<PushRowsResponse> {
const { hostId, hostName, source, rows } = input;
......@@ -99,7 +99,7 @@ export class PushReceiverService {
// 事件驱动:touched 患者入队 persona-recompute(去抖:存量批次重放时窗内合并,见 QueueProducer)
// ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。
// 预检不产生任何副作用 → 不入队(touched 本就为空,此处显式短路,语义更清楚)。
// 预检只保留推送记录,不写业务数据、不入队(touched 本就为空,此处显式短路)。
let personaEnqueued = 0;
for (const { patientId, tenantId } of dryRun ? [] : r.touched) {
try {
......
apps/pac-service/src/modules/sync/push/push.schema.ts
......@@ -82,14 +82,14 @@ export type PushRowsRequest = z.infer<typeof PushRowsRequestSchema>;
export const PushRowsResponseSchema = z.object({
syncLogId: z.string().uuid(),
source: z.string(),
accepted: z.number().int().describe('收到的原生行数'),
transactionsWritten: z.number().int(),
duplicates: z.number().int(),
failed: z.number().int(),
accepted: z.number().int().describe('收到的原生行数,不是成功写入数'),
transactionsWritten: z.number().int().describe('新增流水数,不含直接 upsert;预检时为预计数'),
duplicates: z.number().int().describe('流水幂等命中数,直接 upsert 不计入'),
failed: z.number().int().describe('装配/写入失败计数,按派生资源累计;不能直接忽略'),
/**
* transaction 已落账、但衍生 fact 被 content schema 拒收的条数。
*
* 跟 failed 分开:failed 是"这行没进 PAC";factsFailed 是"行进了,但派生不出可用事实"
* 跟 failed 分开:failed 是装配/写入失败计数;factsFailed 是流水已保存但部分派生事实被拒收。
* (典型:金额为负、必填字段空)。**>0 表示这批数据有一部分对 PAC 无效,要看 factRejects**。
*
* 2026-07-29 补:此前这类拒收既不计数也不回报,回执是 success/failed=0,
......@@ -119,9 +119,9 @@ export const PushRowsResponseSchema = z.object({
/// 宿主自检信号:样本批推完看这两个数,>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 = 本次为预检,未落库'),
/// 预检模式(?dryRun=1):跑 transforms+装配+归一+tenant 解析,不写业务数据、不派生事实,
/// 仍保留预检推送记录。返回预计数,正式推送仍须检查失败计数。
dryRun: z.boolean().describe('true = 预检,未写业务数据,仍保留预检推送记录'),
samples: z
.array(z.object({ resource: z.string(), canonical: z.unknown() }))
.describe('预检样本:各 canonical 资源首行的翻译结果(仅 dryRun 返回,供宿主核对 PAC 是否理解正确)'),
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment